class KafkaClusterObserver implements Closeable {

  public KafkaClusterObserver(String bootstrapServer) {
    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
    properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, String.valueOf(false));
    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
        DESERIALIZER_CLASS);
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
        DESERIALIZER_CLASS);
    this.kafkaConsumer = new KafkaConsumer(properties);
  }

}

class KafkaClusterObserverTest {

  private final KafkaConsumer kafkaConsumer;
  private final KafkaClusterObserver observer;
  private String testTopic1;
  private final List<PartitionInfo> partitionInfo1;
  private String testTopic2;
  private final List<PartitionInfo> partitionInfo2;
  private Map<TopicPartition, Integer> mockPartitionLeaderMap;

  @Test
  public void testKafkaClusterObserver() {

    EasyMock.reset(kafkaConsumer);

    TopicPartition tp1 = new TopicPartition(testTopic1, 1);
    TopicPartition tp2 = new TopicPartition(testTopic2, 1);
    Map<String, List<PartitionInfo>> partitionInfoMap = new HashMap<>();
    partitionInfoMap.put(testTopic1, partitionInfo1);
    partitionInfoMap.put(testTopic2, partitionInfo2);

    EasyMock.expect(kafkaConsumer.listTopics()).andReturn(partitionInfoMap);
    EasyMock.replay(kafkaConsumer);

    Map<TopicPartition, Integer> result =
        observer.findLeaderForPartitions(ImmutableList.of(tp1, tp2));
    Assert.assertEquals(result.get(tp1), Integer.valueOf(1));
    Assert.assertEquals(result.get(tp2), Integer.valueOf(2));

    EasyMock.verify(kafkaConsumer);
  }

}