@Test
public void testAddFetcherForTopicPartition() {

  CustomizedConsumerConfig consumerConfig = new CustomizedConsumerConfig(new Properties());
  consumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "ureplicator");
  consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "ureplicator");
  List<BlockingQueue<FetchedDataChunk>> messageQueue = new ArrayList<>();
  BlockingQueue<FetchedDataChunk> queue = new LinkedBlockingQueue<>(3);
  messageQueue.add(queue);
  MockFetcherManager fetcherManagerByHashId = new MockFetcherManager(
      "FetcherManagerGroupByHashId",
      consumerConfig, messageQueue);
  TopicPartition tp1 = new TopicPartition(testTopic1, 0);
  TopicPartition tp2 = new TopicPartition(testTopic2, 1);
  PartitionOffsetInfo partitionOffsetInfo1 = new PartitionOffsetInfo(tp1, 0L, 10L);
  PartitionOffsetInfo partitionOffsetInfo2 = new PartitionOffsetInfo(tp2, 0L, 10L);
  EasyMock.reset(mockConsumerFetcherThread);
  mockConsumerFetcherThread.start();
  EasyMock.expectLastCall().times(2);
  mockConsumerFetcherThread.addPartitions(Collections.singletonMap(tp1, partitionOffsetInfo1));
  EasyMock.expectLastCall().times(1);
  mockConsumerFetcherThread.addPartitions(Collections.singletonMap(tp2, partitionOffsetInfo2));
  EasyMock.expectLastCall().times(1);
  EasyMock.replay(mockConsumerFetcherThread);
  fetcherManagerByHashId
      .addFetcherForTopicPartition(tp1, partitionOffsetInfo1, "fetcherThread-1");
  fetcherManagerByHashId
      .addFetcherForTopicPartition(tp2, partitionOffsetInfo2, "fetcherThread-2");
  Assert.assertEquals(fetcherManagerByHashId.clientId.size(), 2);
  Assert.assertTrue(fetcherManagerByHashId.clientId.contains("ureplicator-1"));
  Assert.assertTrue(fetcherManagerByHashId.clientId.contains("ureplicator-2"));
  Assert.assertEquals(consumerConfig.getProperty(ConsumerConfig.CLIENT_ID_CONFIG), "ureplicator");
  EasyMock.verify(mockConsumerFetcherThread);
}