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); } KafkaClusterObserver(String bootstrapServer); @VisibleForTesting protected  KafkaClusterObserver(KafkaConsumer kafkaConsumer); Map<TopicPartition, Integer> findLeaderForPartitions(
      List<TopicPartition> topicPartitions); @Override void close();  }
@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); }