@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);
}