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