import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerApp {

  public static void main(String[] args) {
    Properties props = new Properties();
    // We use "put" since there are strings. Otherwise, use setProperty.
    // These are the 3 required properties.
    // 1. Cluster membership: partition leaders, etc
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092, BROKER-1:9092, BROKER-2:9093");

    // 2. How keys and values are deserialized.
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaConsumerCommon.STRING_DESERIALIZER);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaConsumerCommon.STRING_DESERIALIZER);

    // 3. The consumer Group ID is now required.
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "demo");

    KafkaConsumer<String, String> myConsumer = new KafkaConsumer<String, String>(props);

    /* Let the consumer handle the partitions */
    myConsumer.subscribe(Arrays.asList(KafkaCommon.TOPIC));

    try {
      while (true) {
        ConsumerRecords<String, String> records = myConsumer.poll(KafkaConsumerCommon.TIMEOUT);
        KafkaConsumerCommon.process(records);
      }
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      myConsumer.close();
    }
  }

  private static void subscribeExamples(KafkaConsumer<String, String> c) {
    // In tutorial but does not compile: needs a rebalance callback
    // c.subscribe("my-*");
    // Maybe that way ?
    // c.subscribe(Pattern.compile("my-[\\w]+"), null);

    // Typical initial subscription.
    c.subscribe(Arrays.asList(KafkaCommon.TOPIC));

    // Replaces the current subscription set, does not extend it
    c.subscribe(Arrays.asList("another-topic"));

    // Better for incremental cases.
    ArrayList<String> topics = new ArrayList<String>();
    topics.add(KafkaCommon.TOPIC);
    topics.add("my-other-topic");
    topics.add("yetAnotherTopic");
    c.subscribe(topics);

    // Unsubcribe all topics.
    c.unsubscribe();

    // Alternative
    topics.clear();
    c.subscribe(topics);
  }
}