protected List<ReplicaThread> createThreadPool(String datacenter, int numberOfThreads, boolean startThread) {
    nextReplicaThreadIndexByDc.put(datacenter, new AtomicInteger(0));
    List<ReplicaThread> replicaThreads = new ArrayList<>();
    logger.info("Number of replica threads to replicate from {}: {}", datacenter, numberOfThreads);
    ResponseHandler responseHandler = new ResponseHandler(clusterMap);
    replicationMetrics.populateSingleColoMetrics(datacenter);
    for (int i = 0; i < numberOfThreads; i++) {
      boolean replicatingOverSsl = sslEnabledDatacenters.contains(datacenter);
      String threadIdentity = getReplicaThreadName(datacenter, i);
      try {
        StoreKeyConverter threadSpecificKeyConverter = storeKeyConverterFactory.getStoreKeyConverter();
        Transformer threadSpecificTransformer = null;
        if (transformerClassName != null && !transformerClassName.isEmpty()) {
          threadSpecificTransformer =
              Utils.getObj(transformerClassName, storeKeyFactory, threadSpecificKeyConverter);
        }
        ReplicaThread replicaThread =
            getReplicaThread(threadIdentity, tokenHelper, clusterMap, correlationIdGenerator, dataNodeId,
                networkClientFactory.getNetworkClient(), replicationConfig, replicationMetrics, notification,
                threadSpecificKeyConverter, threadSpecificTransformer, metricRegistry, replicatingOverSsl, datacenter,
                responseHandler, time, replicaSyncUpManager, skipPredicate, leaderBasedReplicationAdmin);
        replicaThreads.add(replicaThread);
        createThread(replicaThread, startThread);
      } catch (Exception e) {
        throw new RuntimeException("Encountered exception instantiating ReplicaThread", e);
      }
    }
    replicationMetrics.trackLiveThreadsCount(replicaThreads, datacenter);
    return replicaThreads;
  }
Create thread pool for a datacenter.
@param datacenter The datacenter String.
@param numberOfThreads Number of threads to create for the thread pool.
@param startThread If thread needs to be started when create.
ambry-replication/src/main/java/com/github/ambry/replication/ReplicationEngine.java