/**
 * 独立消费者案例(订阅主题)
 *
 * @author zhongye
 * @since 2022.05.06
 */
public class CustomConsumer {
  public static void main(String[] args) {
    //0 配置
    Properties properties = new Properties();

    //连接bootstrap.servers
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

    // 反序列化
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    // 指定消费者组id
    properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

    // 1 创建一个消费者 "","hello"
    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

    // 2 订阅主题 first
    ArrayList<String> topics = new ArrayList<>();
    topics.add("first");
    kafkaConsumer.subscribe(topics);

    // 3 消费数据
    while (true){
      ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
      for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
      }
    }

  }
}

/**
 * 消费 first 主题下 0 号分区的数据(指定分区消费)。
 *
 * @author zhongye
 * @since 2022.05.06
 */
public class CustomConsumerPartition {
  public static void main(String[] args) {
    //0 配置
    Properties properties = new Properties();

    //连接bootstrap.servers
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

    // 反序列化
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    // 指定消费者组id
    properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

    // 1 创建一个消费者 "","hello"
    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

    // 2 订阅主题对应的分区
    ArrayList<TopicPartition> topics = new ArrayList<>();
    topics.add(new TopicPartition("first",0));
    kafkaConsumer.assign(topics);

    // 3 消费数据
    while (true){
      ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
      for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
      }
    }

  }
}

ConsumerRecord(topic = first, partition = 0, leaderEpoch = 0, offset = 32, CreateTime = 1651840032039, serialized key size = -1, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = silince0)
ConsumerRecord(topic = first, partition = 0, leaderEpoch = 0, offset = 33, CreateTime = 1651840032051, serialized key size = -1, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = silince1)
ConsumerRecord(topic = first, partition = 0, leaderEpoch = 0, offset = 34, CreateTime = 1651840032052, serialized key size = -1, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = silince2)
ConsumerRecord(topic = first, partition = 0, leaderEpoch = 0, offset = 35, CreateTime = 1651840032052, serialized key size = -1, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = silince3)
ConsumerRecord(topic = first, partition = 0, leaderEpoch = 0, offset = 36, CreateTime = 1651840032052, serialized key size = -1, serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = silince4)

// 提交offset的时间周期1000ms，默认5s 
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);

// 是否自动提交offset
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

// 订阅一个主题
ArrayList<String> topics = new ArrayList<>(); 
topics.add("first"); 
kafkaConsumer.subscribe(topics);

Set<TopicPartition> assignment= new HashSet<>();
while (assignment.size() == 0) {  // 保证分区分配方案已经制定完毕
  // 获取消费者分区分配信息(有了分区分配信息才能开始消费) 
  assignment = kafkaConsumer.assignment();
}

// 遍历所有分区，并指定 offset 从 1700 的位置开始消费 
for (TopicPartition tp: assignment) {
  //覆盖使用者将在下一次轮询(超时)中使用的提取偏移量。如果对同一分区多次调用此 API，则将在下一个 poll ()上使用最新的偏移量。
  kafkaConsumer.seek(tp, 1700);
}

/**
 * 指定时间消费
 *
 * @author zhongye
 * @since 2022.05.07
 */
public class CustomConsumerSeekTime {
  public static void main(String[] args) {
    //0 配置
    Properties properties = new Properties();

    //连接bootstrap.servers
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

    // 反序列化
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    // 指定消费者组id
    properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

    // 1 创建一个消费者 "","hello"
    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

    // 2 定义主题 first
    ArrayList<String> topics = new ArrayList<>();
    topics.add("first");
    kafkaConsumer.subscribe(topics);

    // 指定位置进行消费
    Set<TopicPartition> assignment = kafkaConsumer.assignment();

    // 保证分区分配方案已经制定完毕
    while (assignment.size()==0){
      // 获取消费者分区分配信息(有了分区分配信息才能开始消费)
      assignment = kafkaConsumer.assignment();
    }

    // ⭐️希望把时间转换为对应的offset
    HashMap<TopicPartition, Long> topicPartitionLongHashMap = new HashMap<>();
    // 封装对应的map
    for (TopicPartition topicPartition : assignment) {
      // 指定当前时刻一天前的offset
      topicPartitionLongHashMap.put(topicPartition,System.currentTimeMillis()-1*24*3600*1000);
    }
    Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = kafkaConsumer.offsetsForTimes(topicPartitionLongHashMap);

    // 指定消费的offset
    for (TopicPartition topicPartition : assignment) {
      // 获取各分区一天前的offset
      OffsetAndTimestamp offsetAndTimestamp = offsetsForTimes.get(topicPartition);
      kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
    }


    // 3 消费数据
    while (true){
      ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
      for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        System.out.println(consumerRecord);
      }
    }

  }
}