FetcherManager extends ShutdownableThread implements
    IConsumerFetcherManager { @GuardedBy("fetcherMapLock") protected void addFetcherForTopicPartition( TopicPartition tp, PartitionOffsetInfo offsetInfo, String fetcherThreadName) { LOGGER.info("Enter add fetcher thread for partitions {}, fetcherThread {}", tp, fetcherThreadName); synchronized (fetcherMapLock) { if (StringUtils.isBlank(fetcherThreadName)) { LOGGER.warn("Unexpected behavior, can't find threadName for topic partition {}", tp); return; } ConsumerFetcherThread fetcherThread = fetcherThreadMap .getOrDefault(fetcherThreadName, null); if (fetcherThread == null) { try { int queueSize = Math.abs(fetcherThreadName.hashCode() % messageQueue.size()); LOGGER.info("Creating fetcher thread {}", fetcherThreadName); CustomizedConsumerConfig cloned = (CustomizedConsumerConfig) consumerProperties.clone(); String clientIdPrefix = consumerProperties .getProperty(ConsumerConfig.CLIENT_ID_CONFIG, "ureplicator"); int fetcherThreadId = fetcherId.incrementAndGet(); cloned.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, String.format("%s-%d", clientIdPrefix, fetcherThreadId)); fetcherThread = createConsumerFetcherThread(fetcherThreadName, cloned, messageLimiter, messageQueue.get(queueSize)); fetcherThread.start(); fetcherThreadMap.put(fetcherThreadName, fetcherThread); LOGGER.info("Fetcher fetcher thread {} created", fetcherThreadName); } catch (Exception e) { LOGGER.error("Failed to create new fetcher thread {}", getName(), e); return; } } fetcherThread.addPartitions(ImmutableMap.of(tp, offsetInfo)); partitionThreadMap.putIfAbsent(tp, fetcherThreadName); } LOGGER.info("Add fetcher thread for partitions {} finished", tp); } FetcherManager(String threadName,
      CustomizedConsumerConfig consumerProperties,
      List<BlockingQueue<FetchedDataChunk>> messageQueue); @VisibleForTesting protected  FetcherManager(String threadName,
      CustomizedConsumerConfig consumerProperties,
      Map<String, ConsumerFetcherThread> fetcherThreadMap,
      List<BlockingQueue<FetchedDataChunk>> messageQueue); void setMessageRate(Double rate); void addTopicPartition(TopicPartition topicPartition, PartitionOffsetInfo partitionInfo); void removeTopicPartition(TopicPartition topicPartition); void shutdown(); Set<TopicPartition> getTopicPartitions(); @Override void doWork();  }
@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); }