package com.assu.study.chap04.offset;

import java.time.Duration;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;

// 가장 최근의 메시지 배치를 처리한 뒤 commitSync() 를 호출하여 오프셋 커밋
@Slf4j
public class CommitSync {
  public static void main(String[] args) {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountry");

    Duration timeout = Duration.ofMillis(100);

    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);

        for (ConsumerRecord<String, String> record : records) {
          // 여기서는 로그가 출력되면 처리가 끝나는 것으로 간주함
          log.info(
              "topic: {}, partition: {}, offset: {}, customer: {}, country: {}",
              record.topic(),
              record.partition(),
              record.offset(),
              record.key(),
              record.value());
        }

        try {
          // 현재 배치의 모든 레코드에 대한 '처리'가 완료되면 추가 메시지를 폴링하기 전에 commitSync() 를 호출해서
          // 해당 배치의 마지막 오프셋 커밋
          consumer.commitSync();
        } catch (CommitFailedException e) {
          log.error("commit failed: {}", e.getMessage());
        }
      }
    }
  }
}

package com.assu.study.chap04.offset;

import java.time.Duration;
import java.util.Properties;

import lombok.extern.slf4j.Slf4j;
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.common.serialization.StringDeserializer;

@Slf4j
public class CommitAsync {
  public void commitAsync() {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountry");

    Duration timeout = Duration.ofMillis(100);

    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        for (ConsumerRecord<String, String> record : records) {
          // 여기서는 로그가 출력되면 처리가 끝나는 것으로 간주함
          log.info(
              "topic: {}, partition: {}, offset: {}, customer: {}, country: {}",
              record.topic(),
              record.partition(),
              record.offset(),
              record.key(),
              record.value());
        }
        // 마지막 오프셋을 커밋하고 처리 작업을 계속함
        consumer.commitAsync();
      }
    }
  }
}

// 마지막 오프셋을 커밋하고 처리 작업을 계속함
// 하지만, 커밋이 실패할 경우 실패가 났다는 사실과 함께 오프셋을 로그에 남김
consumer.commitAsync(
    new OffsetCommitCallback() {
      @Override
      public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception e) {
        if (e != null) {
          log.error("Commit failed of offsets: {}", e.getMessage());
        }
      }
    });
}

package com.assu.study.chap04.offset;

import java.time.Duration;
import java.util.Map;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

@Slf4j
public class CommitAsync {
  // 컨슈머 종료 직전에 동기적 커밋과 비동기적 커밋 함께 사용하여 마지막 오프셋 커밋
  public void commitSyncAndAsyncBeforeClosingConsumer() {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountry");

    Duration timeout = Duration.ofMillis(100);

    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
      boolean closing = false; // TODO: closing 업데이트
      while (!closing) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        for (ConsumerRecord<String, String> record : records) {
          // 여기서는 로그가 출력되면 처리가 끝나는 것으로 간주함
          log.info(
              "topic: {}, partition: {}, offset: {}, customer: {}, country: {}",
              record.topic(),
              record.partition(),
              record.offset(),
              record.key(),
              record.value());
        }
        // 정상적인 상황에서는 비동기 커밋 사용
        // 더 빠를 뿐더러 커밋이 실패해도 다음 커밋이 성공할 수 있음
        consumer.commitAsync();
      }
      // 컨슈머를 닫는 상황에서는 '다음 커밋' 이 없으므로 commitSync() 를 호출하여
      // 커밋의 성공하면 종료, 회복 불가능한 에러가 발생할 때까지 재시도함
      consumer.commitSync();
    } catch (Exception e) {
      log.error("Unexpected error: {}", e.getMessage());
    }
  }
}

package com.assu.study.chap04.offset;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

@Slf4j
public class CommitAsync {

  int count = 0;
  // 오프셋을 추적하기 위해 사용할 맵
  private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

  // 특정 오프셋 커밋
  public void commitSpecificOffset() {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountry");

    Duration timeout = Duration.ofMillis(100);

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(timeout);
      for (ConsumerRecord<String, String> record : records) {
        // 여기서는 로그가 출력되면 처리가 끝나는 것으로 간주함
        log.info(
            "topic: {}, partition: {}, offset: {}, customer: {}, country: {}",
            record.topic(),
            record.partition(),
            record.offset(),
            record.key(),
            record.value());

        // 각 레코드를 처리한 후 맵을 다음에 처리할 것으로 예상되는 오프셋으로 업데이트
        // 커밋될 오프셋은 애플리케이션이 다음 번에 읽어야 할 메시지의 오프셋이어야 함 (= 향후에 읽기 시작할 메시지의 위치)
        currentOffsets.put(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1, "no metadata"));

        // 10개의 레코드마다 현재 오프셋 커밋
        // 실제 운영 시엔 시간 혹은 레코드의 내용을 기준으로 커밋해야 함
        if (count % 10 == 0) {
          // commitSync() 도 사용 가능
          consumer.commitAsync(currentOffsets, null); // 여기선 callback 을 null 로 처리
        }
        count++;
      }
    }
  }
}

package org.apache.kafka.clients.consumer;

import java.util.Collection;
import org.apache.kafka.common.TopicPartition;

public interface ConsumerRebalanceListener {
  void onPartitionsRevoked(Collection<TopicPartition> partitions);

  void onPartitionsAssigned(Collection<TopicPartition> partitions);

  default void onPartitionsLost(Collection<TopicPartition> partitions) {
    this.onPartitionsRevoked(partitions);
  }
}

package com.assu.study.chap04.rebalance;

import java.time.Duration;
import java.util.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;

// 파티션이 해제되기 직전에 동기적 커밋과 비동기적 커밋 함께 사용하여 마지막 오프셋 커밋
@Slf4j
public class RebalanceListener {
  static Properties props = new Properties();

  static {
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountry");
  }

  private Duration timeout = Duration.ofMillis(100);
  private KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
  private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

  public void commitSyncAndAsyncBeforePartitionRevoked() {
    try {
      // subscribe() 호출 시 ConsumerRebalanceListener 를 인수로 지정하여 컨슈머가 호출할 수 있도록 해줌
      consumer.subscribe(Collections.singletonList("testTopic"), new HandleRebalance());

      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(timeout);
        for (ConsumerRecord<String, String> record : records) {
          log.info(
              "topic: {}, partition: {}, offset: {}, customer: {}, country: {}",
              record.topic(),
              record.partition(),
              record.offset(),
              record.key(),
              record.value());

          currentOffsets.put(
              new TopicPartition(record.topic(), record.partition()),
              new OffsetAndMetadata(record.offset() + 1, null));
          consumer.commitAsync(currentOffsets, null); // 여기선 callback 을 null 로 처리
        }
      }
    } catch (WakeupException e) {
      // ignore, we're closing
    } catch (Exception e) {
      log.error("Unexpected error: {}", e.getMessage());
    } finally {
      try {
        consumer.commitSync(currentOffsets);
      } finally {
        consumer.close();
        log.info("Closed consumer and we are done.");
      }
    }
  }

  // ConsumerRebalanceListener 구현
  private class HandleRebalance implements ConsumerRebalanceListener {
    // 여기서는 컨슈머가 새 파티션을 할당받았을 때 아무것도 할 필요가 없기 때문에 바로 메시지를 읽기 시작함
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {}

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
      log.error("Lost partitions in rebalance. Commiting current offsets: {}", currentOffsets);
      // 리밸런싱 때문에 파티션이 해제될 상황이라면 오프셋 커밋
      // 할당 해제될 파티션의 오프셋 뿐 아니라 모든 파티션에 대한 오프셋 커밋
      // (이 오프셋들은 이미 처리된 이벤트들의 오프셋이므로 문제없음)
      // 리밸런스가 진행되기 전에 모든 오프셋이 확실히 커밋되도록 commitSync() 사용
      consumer.commitSync(currentOffsets);
    }
  }
}

package com.assu.study.chap04.offset;

import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.util.Map;
import java.util.Properties;
import java.util.stream.Collectors;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

public class Seek {
  // 모든 파티션의 현재 오프셋을 특정한 시각에 생성된 레코드의 오프셋으로 설정
  public void setOffset() {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountry");

    Duration timeout = Duration.ofMillis(100);
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

    Long oneHourEarlier =
        Instant.now().atZone(ZoneId.systemDefault()).minusHours(1).toEpochSecond();

    // consumer.assignment() 로 얻어온 컨슈머에 할당된 모든 파티션에 대해
    // 컨슈머를 되돌리고자 하는 타임스탬프 값을 담은 맵 생성
    Map<TopicPartition, Long> partitionTimestampMap =
        consumer.assignment().stream().collect(Collectors.toMap(tp -> tp, tp -> oneHourEarlier));

    // 각 타임스탬프에 대한 오프셋 받아옴
    // offsetsForTimes() 는 브로커에 요청을 보내서 타임스탬프 인덱스에 저장된 오프셋을 리턴함
    Map<TopicPartition, OffsetAndTimestamp> offsetMap = consumer.offsetsForTimes(partitionTimestampMap);

    // 각 파티션의 오프셋을 앞 단계에서 리턴된 오프셋으로 재설정
    for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry: offsetMap.entrySet()) {
      consumer.seek(entry.getKey(), entry.getValue().offset());
    }
  }
}