package com.assu.study.chap04_simplemovingavg.newconsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.queue.CircularFifoQueue;
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.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;

// 메인 애플리케이션 스레드에서 돌아가는 컨슈머의 실행 루프 종료
@Slf4j
public class SimpleMovingAvgNewConsumer {
  private Properties kafkaProps = new Properties();
  private String waitTime;
  private KafkaConsumer<String, String> consumer;

  public static void main(String[] args) {
    if (args.length == 0) {
      log.error("need args: {brokers} {group.id} {topic} {window-size}");
      return;
    }

    final SimpleMovingAvgNewConsumer movingAvg = new SimpleMovingAvgNewConsumer();
    String brokers = args[0];
    String groupId = args[1];
    String topic = args[2];
    int windowSize = Integer.parseInt(args[3]);

    CircularFifoQueue queue = new CircularFifoQueue(windowSize);
    movingAvg.configure(brokers, groupId);

    final Thread mainThread = Thread.currentThread();

    // ShutdownHook 은 별개의 스레드에서 실행되므로 폴링 루프를 탈출하기 위해 할 수 있는 것은 wakeup() 을 호출하는 것 뿐임
    Runtime.getRuntime()
        .addShutdownHook(
            new Thread() {
              @Override
              public void run() {
                log.info("Starting exit...");
                movingAvg.consumer.wakeup();

                try {
                  mainThread.join();
                } catch (InterruptedException e) {
                  log.error("error: {}", e.getMessage());
                }
              }
            });

    try {
      movingAvg.consumer.subscribe(Collections.singletonList(topic));

      Duration timeout = Duration.ofMillis(10000);  // 10초

      //  Ctrl+c 가 눌리면 shutdownhook 정리 및 종료
      while (true) {
        // 타임아웃을 매우 길게 설정함
        // 만일 폴링 루프가 충분히 짧아서 종료되기 전에 좀 기다리는게 별로 문제가 되지 않는다면 굳이 wakeup() 을 호출해줄 필요가 없음
        // 즉, 그냥 이터레이션마다 아토믹 boolean 값을 확인하는 것만으로 충분함
        // 폴링 타임아웃을 길게 잡아주는 이유는 메시지가 조금씩 쌓이는 토픽에서 데이터를 읽어올 때 편리하기 때문임
        // 이 방법을 사용하면 브로커가 리턴할 새로운 데이터를 가지고 있지 않은 동안 계속해서 루프를 돌면서도 더 적은 CPU 를 사용할 수 있음
        ConsumerRecords<String, String> records = movingAvg.consumer.poll(timeout);
        log.info("{} -- waiting for data...", System.currentTimeMillis());

        for (ConsumerRecord<String, String> record : records) {
          log.info(
              "topic: {}, partition: {}, offset: {}, key: {}, value: {}",
              record.topic(),
              record.partition(),
              record.offset(),
              record.key(),
              record.value());

          int sum = 0;

          try {
            int num = Integer.parseInt(record.value());
            queue.add(num);
          } catch (NumberFormatException e) {
            // just ignore strings
          }

          for (Object o : queue) {
            sum += (Integer) o;
          }

          if (queue.size() > 0) {
            log.info("Moving avg is {}", (sum / queue.size()));
          }
        }
        for (TopicPartition tp : movingAvg.consumer.assignment()) {
          log.info("Committing offset at position: {}", movingAvg.consumer.position(tp));
        }

        movingAvg.consumer.commitSync();
      }
    } catch (WakeupException e) {
      // ignore for shutdown
      // 다른 스레드에서 wakeup() 을 호출할 경우, 폴링 루프에서 WakeupException 발생
      // 발생된 예외를 잡아줌으로써 애플리케이션이 예기치않게 종료되지 않도록 할 수 있지만 딱히 뭔가를 추가적으로 해 줄 필요ㅇ는 없음
    } finally {
      // 컨슈머 종료 전 닫아서 리소스 정리
      movingAvg.consumer.close();
      log.info("Closed consumer and done.");
    }
  }

  private void configure(String servers, String groupId) {
    kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, servers);
    kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
    // 유효한 오프셋이 없을 경우 파티션의 맨 처음부터 모든 데이터 읽음
    kafkaProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

    consumer = new KafkaConsumer<>(kafkaProps);
  }
}

// org.apache.kafka.common.serialization.StringSerializer 리턴
Serializer<String> serializer = Serdes.String().serializer();

// org.apache.kafka.common.serialization.StringDeserializer 리턴
Deserializer<String> deserializer = Serdes.String().deserializer();

kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Serdes.String().deserializer().getClass().getName());
kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, Serdes.String().deserializer().getClass().getName());

package com.assu.study.chap04.deserializer;

import lombok.Getter;
import lombok.RequiredArgsConstructor;

@RequiredArgsConstructor
@Getter
public class Customer {
  private final int customerId;
  private final String customerName;
}

package com.assu.study.chap04.deserializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

// Customer 클래스를 위한 커스텀 디시리얼라이저

// 컨슈머에서도 Customer 클래스 필요
// 프로듀서에 사용되는 클래스 및 시리얼라이저는 컨슈머 애플리케이션에서도 같은 것이 사용되어야 함
// 같은 데이터를 공유해서 사용하는 컨슈머와 프로듀서의 수가 많은 조직에서는 꽤나 어려운 일임
public class CustomerDeserializer implements Deserializer<Customer> {
  @Override
  public Customer deserialize(String topic, byte[] data) {
    int id;
    int nameSize;
    String name;

    try {
      if (data == null) {
        return null;
      }

      if (data.length < 8) {
        throw new SerializationException(
            "Size of data received by deserializer is shorter than expected.");
      }

      ByteBuffer buf = ByteBuffer.wrap(data);
      id = buf.getInt();
      nameSize = buf.getInt();

      byte[] nameBytes = new byte[nameSize];
      buf.get(nameBytes);
      name = new String(nameBytes, StandardCharsets.UTF_8);

      // 고객 id 와 이름을 바이트 배열로부터 꺼내온 후 필요로 하는 객체 생성
      return new Customer(id, name);
    } catch (Exception e) {
      throw new SerializationException("Error when deserializing byte[] to Customer: ", e);
    }
  }

  @Override
  public void configure(Map<String, ?> configs, boolean isKey) {
    // nothing to configure
  }

  @Override
  public void close() {
    // nothing to close
  }
}

package com.assu.study.chap04.deserializer;

import java.time.Duration;
import java.util.Collections;
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.Serdes;

@Slf4j
public class ConsumerUseCustomerDeserializer {
  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,
        Serdes.String().deserializer().getClass().getName());
    // CustomerDeserializer 클래스 설정
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CustomerDeserializer.class.getName());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "CustomerCountry");

    Duration timeout = Duration.ofMillis(100);

    KafkaConsumer<String, Customer> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList("topic"));

    while (true) {
      ConsumerRecords<String, Customer> records = consumer.poll(timeout);
      for (ConsumerRecord<String, Customer> record : records) {
        log.info(
            "current customer Id: {}, customer name: {}",
            record.value().getCustomerId(),
            record.value().getCustomerName());
      }
      consumer.commitSync();
    }
  }
}

package com.assu.study.chap04.avrosample;

import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import javasessionize.avro.LogLine;
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.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.Serdes;

@Slf4j
public class AvroConsumer {
  private final KafkaConsumer<String, LogLine> consumer;
  private final KafkaProducer<String, LogLine> producer;
  private final String inputTopic;
  private final String outputTopic;
  private Map<String, SessionState> state = new HashMap<>();

  public AvroConsumer(
      String brokers, String groupId, String inputTopic, String outputTopic, String url) {
    this.consumer = new KafkaConsumer<>(createConsumerConfig(brokers, groupId, url));
    this.producer = getProducer(outputTopic, url);
    this.inputTopic = inputTopic;
    this.outputTopic = outputTopic;
  }

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    // 하드 코딩
    String groupId = "avroSample";
    String inputTopic = "iTopic";
    String outputTopic = "oTopic";
    String url = "http://localhost:8081";
    String brokers = "localhost:9092";

    AvroConsumer avroConsumer = new AvroConsumer(brokers, groupId, inputTopic, outputTopic, url);

    avroConsumer.run();
  }

  private void run() throws ExecutionException, InterruptedException {
    consumer.subscribe(Collections.singletonList(inputTopic));

    log.info("Reading topic: {}", inputTopic);

    Duration timeout = Duration.ofMillis(1000);

    while (true) {
      // 레코드 밸류 타입으로 LogLine 지정
      ConsumerRecords<String, LogLine> records = consumer.poll(timeout);
      for (ConsumerRecord<String, LogLine> record : records) {
        String ip = record.key();
        LogLine event = record.value();

        SessionState oldState = state.get(ip);
        int sessionId = 0;

        if (oldState == null) {
          state.put(ip, new SessionState(event.getTimestamp(), 0));
        } else {
          sessionId = oldState.getSessionId();

          // old timestamp 가 30분 이전의 것이면 새로운 세션 생성
          if (oldState.getLastConnection() < event.getTimestamp() - (30 * 60 * 1000)) {
            sessionId = sessionId + 1;
          }
          SessionState newState = new SessionState(event.getTimestamp(), sessionId);
          state.put(ip, newState);
        }

        event.setSessionid(sessionId);

        log.info("event: {}", event);

        ProducerRecord<String, LogLine> sessionizedEvent =
            new ProducerRecord<>(outputTopic, event.getIp().toString(), event);
        producer.send(sessionizedEvent).get();
      }
      consumer.commitSync();
    }
  }

  private Properties createConsumerConfig(String brokers, String groupId, String url) {
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    // 스키마가 저장된 곳
    // 프로듀서가 등록한 스키마를 컨슈머가 역직렬화하기 위해 사용
    props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, url);
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
    props.put(
        ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
        Serdes.String().deserializer().getClass().getName());
    // Avro 메시지를 역직렬화하기 위해 KafkaAvroDeserializer 사용
    props.put(
        ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());

    return props;
  }

  private KafkaProducer<String, LogLine> getProducer(String topic, String url) {
    Properties props = new Properties();
    // 여기선 카프카 서버 URI 하드코딩
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    props.put(ProducerConfig.RETRIES_CONFIG, 0);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
    props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, url);

    KafkaProducer<String, LogLine> producer = new KafkaProducer<>(props);
    return producer;
  }
}

package com.assu.study.chap04.avrosample;

import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;

@RequiredArgsConstructor
@Getter
@Setter
public class SessionState {
  private final long lastConnection;
  private final int sessionId;
}

package com.assu.study.chap04.list;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.serialization.*;

// List<T> 에서 키 값에 대한 시리얼라이저, 디시리얼라이저
public class ListSerializerSample {
  // List<T> 키 값에 대한 시리얼라이저
  public void keySerializer() {
    // ListSerializer 객체 생성
    ListSerializer<?> listSerializer = new ListSerializer<>();

    // 생성한 객체 설정
    Map<String, Object> props = new HashMap<>();

    // 밸류값의 경우엔 default.list.value.serde.inner
    props.put(
        CommonClientConfigs.DEFAULT_LIST_KEY_SERDE_INNER_CLASS, Serdes.StringSerde.class.getName());

    // 밸류값의 경우엔 두 번째 인수가 false
    listSerializer.configure(props, true);

    // 설정된 시리얼라이저 얻어옴
    final Serializer<?> inner = listSerializer.getInnerSerializer();
  }

  // List<T> 키 값에 대한 디시리얼라이저
  public void keyDeserializer() {
    // ListDeserializer 객체 생성
    ListDeserializer<?> listDeserializer = new ListDeserializer<>();

    // 생성한 객체 설정
    Map<String, Object> props = new HashMap<>();

    // 바이트 뭉치를 디시리얼라이즈 한 뒤 ArrayList 객체 형태로 리턴
    // LinkedList.class.getName() 으로 설정해주면 LinkedList 객체가 리턴됨
    props.put(CommonClientConfigs.DEFAULT_LIST_KEY_SERDE_TYPE_CLASS, ArrayList.class.getName());

    // 밸류값의 경우엔 default.list.value.serde.inner
    props.put(
        CommonClientConfigs.DEFAULT_LIST_KEY_SERDE_INNER_CLASS, Serdes.StringSerde.class.getName());

    // 밸류값의 경우엔 두 번째 인수가 false
    listDeserializer.configure(props, true);

    // 설정된 디시리얼라이저 얻어옴
    final Deserializer<?> inner = listDeserializer.innerDeserializer();
  }
}

// LinkedList.class.getName() 으로 설정해주면 LinkedList 객체가 리턴됨
props.put(CommonClientConfigs.DEFAULT_LIST_KEY_SERDE_TYPE_CLASS, ArrayList.class.getName());

package com.assu.study.chap04.standalone;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
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.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

// 컨슈머 그룹없이 컨슈머 스스로가 특정 토픽의 모든 파티션을 할당한 뒤 메시지를 읽고 처리
@Slf4j
public class StandaloneConsumer {
  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, "CountryCounter");

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    List<TopicPartition> partitions = new ArrayList<>();
    // 카프카 클러스터에 해당 토픽에 대해 사용 가능한 파티션들을 요청
    // 만일 특정 파티션의 레코드만 읽어올거면 생략해도 됨
    List<PartitionInfo> partitionInfos = consumer.partitionsFor("topic");

    if (partitions != null) {
      for (PartitionInfo partitionInfo : partitionInfos) {
        partitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
      }

      // 읽고자 하는 파티션이 있다면 해당 목록에 `assign()` 으로 추가
      consumer.assign(partitions);

      Duration timeout = Duration.ofMillis(100);

      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.commitSync();
      }
    }
  }
}