public void startup() throws InstantiationException {
    try {
      logger.info("starting");
      clusterParticipants = clusterAgentsFactory.getClusterParticipants();
      if (clusterParticipants == null || clusterParticipants.isEmpty()) {
        logger.info("Cluster participant list is null or empty");
      }
      ClusterParticipant clusterParticipant = clusterParticipants == null || clusterParticipants.isEmpty() ? null : clusterParticipants.get(0);
      logger.info("Setting up JMX.");

      long startTime = SystemTime.getInstance().milliseconds();
      reporter = reporterFactory != null ? reporterFactory.apply(registry) : JmxReporter.forRegistry(registry).build();
      reporter.start();

      logger.info("creating configs");
      NetworkConfig networkConfig = new NetworkConfig(properties);
      StoreConfig storeConfig = new StoreConfig(properties);
      DiskManagerConfig diskManagerConfig = new DiskManagerConfig(properties);
      ServerConfig serverConfig = new ServerConfig(properties);
      ReplicationConfig replicationConfig = new ReplicationConfig(properties);
      ConnectionPoolConfig connectionPoolConfig = new ConnectionPoolConfig(properties);
      SSLConfig sslConfig = new SSLConfig(properties);
      StatsManagerConfig statsConfig = new StatsManagerConfig(properties);
      FileCopyBasedReplicationConfig fileCopyBasedReplicationConfig = new FileCopyBasedReplicationConfig(properties);
      ReplicaPrioritizationConfig replicaPrioritizationConfig = new ReplicaPrioritizationConfig(properties);
      // verify the configs
      properties.verify();

      AccountServiceFactory accountServiceFactory =
          Utils.getObj(serverConfig.serverAccountServiceFactory, properties, registry);
      AccountService accountService = accountServiceFactory.getAccountService();

      SSLFactory sslHttp2Factory = new NettySslHttp2Factory(sslConfig);
      StoreKeyConverterFactory storeKeyConverterFactory =
          Utils.getObj(serverConfig.serverStoreKeyConverterFactory, properties, registry);

      if (serverConfig.serverExecutionMode.equals(ServerExecutionMode.DATA_RECOVERY_MODE)) {
        logger.info("Server execution mode is DATA_RECOVERY_MODE");
        /**
         * Recovery from cloud. When the server is restoring a backup from cloud, it will not replicate from peers.
         * We need to use RecoveryManager for one reason. It will add one replica for the server to replicate from
         * which is the Cloud. The regular ReplicationManager will add 3 replicas, and as a result 3 threads will try
         * to fetch the same data from Azure. Although this can be avoided with a minor change to ReplicationManager,
         * having a separate RecoveryManager for cloud offers the flexibility to override methods and add more
         * suited to cloud.
         */
        StoreKeyFactory storeKeyFactory = Utils.getObj(storeConfig.storeKeyFactory, clusterMap);
        DataNodeId nodeId = clusterMap.getDataNodeId(networkConfig.hostName, networkConfig.port);
        if (nodeId == null) {
          throw new IllegalArgumentException(String.format("Node %s absent in cluster-map", networkConfig.hostName));
        }
        // In most cases, there should be only one participant in the clusterParticipants list. If there are more than one
        // and some components require sole participant, the first one in the list will be primary participant.
        scheduler = Utils.newScheduler(1, true);
        storageManager =
            new StorageManager(storeConfig, diskManagerConfig, scheduler, registry, storeKeyFactory, clusterMap, nodeId,
                new BlobStoreHardDelete(), clusterParticipants, time, new BlobStoreRecovery(), accountService);
        storageManager.start();
        networkClientFactory =
            new RecoveryNetworkClientFactory(properties, registry, clusterMap, storageManager, accountService);
        recoveryManager =
            new RecoveryManager(replicationConfig, clusterMapConfig, storeConfig, storageManager, storeKeyFactory,
                clusterMap, scheduler, nodeId, networkClientFactory, registry, notificationSystem,
                storeKeyConverterFactory, serverConfig.serverMessageTransformer, null, null);
        recoveryManager.start();
      } else if (serverConfig.serverExecutionMode.equals(ServerExecutionMode.DATA_VERIFICATION_MODE)) {
        logger.info("Server execution mode is DATA_VERIFICATION_MODE");
        /**
         * We need composite cluster manager because it has both static and helix cluster manager.
         * We use helix-cluster-map to discover replicas of a partition.
         * We use static-cluster-map to discover local disks on the host.
         * The static map returns objects used for static maps. We need to convert them to be suitable for helix map.
         */
        if (!(clusterMap instanceof CompositeClusterManager)) {
          throw new InstantiationException("Cluster manager must be of type CompositeClusterManager");
        }
        CompositeClusterManager compositeClusterManager = (CompositeClusterManager) clusterMap;
        HelixClusterManager helixClusterManager = compositeClusterManager.getHelixClusterManager();
        StaticClusterManager staticClusterManager = compositeClusterManager.getStaticClusterManager();
        DataNodeId nodeId = staticClusterManager.getDataNodeId(networkConfig.hostName, networkConfig.port);
        if (nodeId == null) {
          throw new IllegalArgumentException(String.format("Node %s absent in cluster-map", networkConfig.hostName));
        }
        // Store key factory is instantiated in two places : here and in internal fork. Careful!
        StoreKeyFactory storeKeyFactory = Utils.getObj(storeConfig.storeKeyFactory, helixClusterManager);
        scheduler = Utils.newScheduler(1, true);
        storageManager =
            new StorageManager(storeConfig, diskManagerConfig, scheduler, registry, storeKeyFactory, staticClusterManager, nodeId,
                new BlobStoreHardDelete(), clusterParticipants, time, new BlobStoreRecovery(), accountService);
        storageManager.start();
        /**
         * Backup integrity monitor here because vcr does not have code to store to disk. Only server does.
         * DataNodeId -> AmbryDataNode -> AmbryServerDataNode : for helix
         * DataNodeId -> DataNode : for static
         */
        AmbryServerDataNode helixNode = new AmbryServerDataNode(nodeId, clusterMapConfig);
        recoveryManager =
            new RecoveryManager(replicationConfig, clusterMapConfig, storeConfig, storageManager, storeKeyFactory,
                helixClusterManager, null, helixNode,
                new RecoveryNetworkClientFactory(properties, registry, helixClusterManager, storageManager, accountService),
                registry, null, storeKeyConverterFactory, serverConfig.serverMessageTransformer,
                null, null);
        replicationManager =
            new ReplicationManager(replicationConfig, clusterMapConfig, storeConfig, storageManager, storeKeyFactory,
                helixClusterManager, null, helixNode,
                new Http2NetworkClientFactory(new Http2ClientMetrics(registry), new Http2ClientConfig(properties), sslHttp2Factory, time),
                registry, null, storeKeyConverterFactory, serverConfig.serverMessageTransformer,
                null, null);
        backupIntegrityMonitor = new BackupIntegrityMonitor(recoveryManager, replicationManager,
            compositeClusterManager, storageManager, helixNode, properties);
        backupIntegrityMonitor.start();
      } else if (serverConfig.serverExecutionMode.equals(ServerExecutionMode.DATA_SERVING_MODE)) {
        logger.info("Server execution mode is DATA_SERVING_MODE");
        scheduler = Utils.newScheduler(serverConfig.serverSchedulerNumOfthreads, false);
        accountStatsMySqlStore =
            statsConfig.enableMysqlReport ? (AccountStatsMySqlStore) new AccountStatsMySqlStoreFactory(properties,
                clusterMapConfig, registry).getAccountStatsStore() : null;

        StoreKeyFactory storeKeyFactory = Utils.getObj(storeConfig.storeKeyFactory, clusterMap);
        FindTokenHelper findTokenHelper = new FindTokenHelper(storeKeyFactory, replicationConfig);
        Callback<AggregatedAccountStorageStats> accountServiceCallback = new AccountServiceCallback(accountService);

        logger.info("checking if node exists in clustermap host {} port {}", networkConfig.hostName, networkConfig.port);
        DataNodeId nodeId = clusterMap.getDataNodeId(networkConfig.hostName, networkConfig.port);
        // Flow for new node ADD
        // Instance config is populated with auto-registration
        // Once DataNode config is populated, Storage, Replication and Stats Manager are created
        // State Machine Model and other tasks are not registered until then to the HelixParticipant
        List<AmbryStatsReport> ambryStatsReports = getAmbryStatsReports(serverConfig);

        for (ClusterParticipant participant : clusterParticipants) {
          participant.participateAndBlockStateTransition(ambryStatsReports, accountStatsMySqlStore, accountServiceCallback);
        }

        // wait for dataNode to be populated
        if (nodeId == null) {
          logger.info("Waiting on dataNode config to be populated...");
          if(!dataNodeLatch.await(serverConfig.serverDatanodeConfigTimeout, TimeUnit.SECONDS)) {
            throw new IllegalArgumentException("Startup timed out waiting for data node config to be populated");
          }
          nodeId = clusterMap.getDataNodeId(networkConfig.hostName, networkConfig.port);
          if (nodeId == null) {
            throw new IllegalArgumentException(String.format("Node %s absent in cluster-map", networkConfig.hostName));
          }
          logger.info("DataNode config is populated");
        }

        // In most cases, there should be only one participant in the clusterParticipants list. If there are more than one
        // and some components require sole participant, the first one in the list will be primary participant.
        logger.info("Creating Storage Manager");
        storageManager =
            new StorageManager(storeConfig, diskManagerConfig, scheduler, registry, storeKeyFactory, clusterMap, nodeId,
                new BlobStoreHardDelete(), clusterParticipants, time, new BlobStoreRecovery(), accountService);
        storageManager.start();

        logger.info("Creating StatsManager to publish stats");
        statsManager =
            new StatsManager(storageManager, clusterMap, clusterMap.getReplicaIds(nodeId),
                registry, statsConfig, time, clusterParticipant, accountStatsMySqlStore, accountService, nodeId);
        statsManager.start();

        // if there are more than one participant on local node, we create a consistency checker to monitor and alert any
        // mismatch in sealed/stopped replica lists that maintained by each participant.
        if (clusterParticipants != null && clusterParticipants.size() > 1
            && serverConfig.serverParticipantsConsistencyCheckerPeriodSec > 0) {
          consistencyChecker = new ParticipantsConsistencyChecker(clusterParticipants, metrics);
          logger.info("Scheduling participants consistency checker with a period of {} secs",
              serverConfig.serverParticipantsConsistencyCheckerPeriodSec);
          consistencyCheckerScheduler = Utils.newScheduler(1, "consistency-checker-", false);
          consistencyCheckerTask = consistencyCheckerScheduler.scheduleAtFixedRate(consistencyChecker, 0,
              serverConfig.serverParticipantsConsistencyCheckerPeriodSec, TimeUnit.SECONDS);
        }

        if (clusterMapConfig.clusterMapEnableHttp2Replication) {
          Http2ClientMetrics http2ClientMetrics = new Http2ClientMetrics(registry);
          Http2ClientConfig http2ClientConfig = new Http2ClientConfig(properties);
          networkClientFactory =
              new Http2NetworkClientFactory(http2ClientMetrics, http2ClientConfig, sslHttp2Factory, time);
          connectionPool = new Http2BlockingChannelPool(sslHttp2Factory, http2ClientConfig, http2ClientMetrics);
        } else {
          SSLFactory sslSocketFactory =
              clusterMapConfig.clusterMapSslEnabledDatacenters.length() > 0 ? SSLFactory.getNewInstance(sslConfig) : null;
          networkClientFactory =
              new SocketNetworkClientFactory(new NetworkMetrics(registry), networkConfig, sslSocketFactory, 20, 20, 50000,
                  time);
          connectionPool = new BlockingChannelConnectionPool(connectionPoolConfig, sslConfig, clusterMapConfig, registry);
        }
        Predicate<MessageInfo> skipPredicate = new ReplicationSkipPredicate(accountService, replicationConfig);
        replicationManager =
            new ReplicationManager(replicationConfig, clusterMapConfig, storeConfig, storageManager, storeKeyFactory,
                clusterMap, scheduler, nodeId, networkClientFactory, registry, notificationSystem,
                storeKeyConverterFactory, serverConfig.serverMessageTransformer, clusterParticipant,
                skipPredicate);
        replicationManager.start();

        if(serverConfig.serverReplicationProtocolForHydration.equals(ServerReplicationMode.FILE_BASED)) {
          FileCopyHandlerFactory fileCopyHandlerFactory =
              new StoreFileCopyHandlerFactory(connectionPool, storageManager, clusterMap,
                  fileCopyBasedReplicationConfig, storeConfig);

          PrioritizationManagerFactory prioritizationManagerFactory = new FileBasedReplicationPrioritizationManagerFactory();
          prioritizationManager = prioritizationManagerFactory.getPrioritizationManager(replicaPrioritizationConfig.replicaPrioritizationStrategy);
          prioritizationManager.start();
          FileCopyBasedReplicationSchedulerFactory fileCopyBasedReplicationSchedulerFactory =
              new FileCopyBasedReplicationSchedulerFactoryImpl(fileCopyHandlerFactory, fileCopyBasedReplicationConfig,
                  clusterMap, prioritizationManager, storageManager, storeConfig, nodeId, clusterParticipant);
          fileCopyBasedReplicationManager =
              new FileCopyBasedReplicationManager(fileCopyBasedReplicationConfig, clusterMapConfig, storageManager,
                  clusterMap, networkClientFactory, new MetricRegistry(), clusterParticipant,
                  fileCopyBasedReplicationSchedulerFactory, fileCopyHandlerFactory, prioritizationManager, storeConfig,
                  replicaPrioritizationConfig);
          fileCopyBasedReplicationManager.start();
        }
        // unblock state transition
        logger.info("Unblocking state transition");
        for (ClusterParticipant participant : clusterParticipants) {
          participant.unblockStateTransition();
        }

        ArrayList<Port> ports = new ArrayList<Port>();
        ports.add(new Port(networkConfig.port, PortType.PLAINTEXT));
        if (nodeId.hasSSLPort()) {
          ports.add(new Port(nodeId.getSSLPort(), PortType.SSL));
        }
        networkServer = new SocketServer(networkConfig, sslConfig, registry, ports);
        requests = new AmbryServerRequests(storageManager, networkServer.getRequestResponseChannel(), clusterMap, nodeId,
            registry, metrics, findTokenHelper, notificationSystem, replicationManager, storeKeyFactory, serverConfig,
            diskManagerConfig, storeKeyConverterFactory, statsManager, clusterParticipant);
        requestHandlerPool = new RequestHandlerPool(serverConfig.serverRequestHandlerNumSocketServerThreads,
            networkServer.getRequestResponseChannel(), requests);
        networkServer.start();

        // Start netty http2 server
        if (nodeId.hasHttp2Port()) {
          NettyConfig nettyConfig = new NettyConfig(properties);
          NettyMetrics nettyMetrics = new NettyMetrics(registry);
          Http2ServerMetrics http2ServerMetrics = new Http2ServerMetrics(registry);
          Http2ClientConfig http2ClientConfig = new Http2ClientConfig(properties);

          logger.info("Http2 port {} is enabled. Starting HTTP/2 service.", nodeId.getHttp2Port());
          NettyServerRequestResponseChannel requestResponseChannel =
              new NettyServerRequestResponseChannel(networkConfig, http2ServerMetrics, metrics,
                  new ServerRequestResponseHelper(clusterMap, findTokenHelper));
          AmbryServerRequests ambryServerRequestsForHttp2 =
              new AmbryServerRequests(storageManager, requestResponseChannel, clusterMap, nodeId, registry, metrics,
                  findTokenHelper, notificationSystem, replicationManager, storeKeyFactory, serverConfig,
                  diskManagerConfig, storeKeyConverterFactory, statsManager, clusterParticipant, connectionPool);
          requestHandlerPoolForHttp2 =
              new RequestHandlerPool(serverConfig.serverRequestHandlerNumOfThreads, requestResponseChannel,
                  ambryServerRequestsForHttp2);

          NioServerFactory nioServerFactory =
              new StorageServerNettyFactory(nodeId.getHttp2Port(), requestResponseChannel, sslHttp2Factory, nettyConfig,
                  http2ClientConfig, metrics, nettyMetrics, http2ServerMetrics, serverSecurityService);
          nettyHttp2Server = nioServerFactory.getNioServer();
          nettyHttp2Server.start();
        }

        if (serverConfig.serverRepairRequestsDbFactory != null) {
          // if we have a RepairRequestsDB, start the threads to fix the partially failed requests.
          try {
            RepairRequestsDbFactory factory =
                Utils.getObj(serverConfig.serverRepairRequestsDbFactory, properties, registry, nodeId.getDatacenterName(),
                    time);
            repairRequestsDb = factory.getRepairRequestsDb();

            localChannel = new LocalRequestResponseChannel();
            LocalNetworkClientFactory localClientFactory =
                new LocalNetworkClientFactory(localChannel, networkConfig, new NetworkMetrics(registry), time);
            AmbryRequests repairRequests =
                new AmbryServerRequests(storageManager, localChannel, clusterMap, nodeId, registry, metrics,
                    findTokenHelper, notificationSystem, replicationManager, storeKeyFactory, serverConfig,
                    diskManagerConfig, storeKeyConverterFactory, statsManager, clusterParticipant,
                    connectionPool);
            // Right now we only open single thread for the repairHandlerPool
            repairHandlerPool = new RequestHandlerPool(1, localChannel, repairRequests, "Repair-");
            // start the repairRequestSender. It sends requests through the localChannel to the repairHandlerPool
            repairRequestsSender =
                new RepairRequestsSender(localChannel, localClientFactory, clusterMap, nodeId, repairRequestsDb,
                    clusterParticipant, registry, storageManager);
            repairThread = Utils.daemonThread("Repair-Sender", repairRequestsSender);
            repairThread.start();
            logger.info("RepairRequests: open the db and started the handling thread {}.", repairRequestsDb);
          } catch (Exception e) {
            logger.error("RepairRequests: Cannot connect to the RepairRequestsDB. ", e);
          }
        }

        if (replicationConfig.enableReplicationPrioritization && clusterMap instanceof HelixClusterManager) {
          HelixClusterManager helixClusterManager = (HelixClusterManager) clusterMap;
          DisruptionServiceFactory disruptionServiceFactory = Utils.getObj(replicationConfig.disruptionServiceFactory, properties, nodeId.getDatacenterName());
          DisruptionService disruptionService = disruptionServiceFactory.getDisruptionService();
          ScheduledExecutorService scheduledExecutorService = Utils.newScheduler(1, "ambry-prioritization", false);
          replicationPrioritizationManager = new ReplicationPrioritizationManager(replicationManager, clusterMap, nodeId, scheduledExecutorService, storageManager, replicationConfig,
          helixClusterManager.getManagerQueryHelper(), disruptionService, registry);
        }

      } else {
        throw new IllegalArgumentException("Unknown server execution mode");
      }

      if (nettyInternalMetrics != null) {
        nettyInternalMetrics.start();
        logger.info("NettyInternalMetric starts");
      }
      logger.info("started");
      long processingTime = SystemTime.getInstance().milliseconds() - startTime;
      metrics.serverStartTimeInMs.update(processingTime);
      logger.info("Server startup time in Ms {}", processingTime);
    } catch (Exception e) {
      logger.error("Error during startup", e);
      throw new InstantiationException("failure during startup " + e);
    }
  }
@param properties {@link VerifiableProperties} object containing configs for the server.
@param clusterAgentsFactory The {@link ClusterAgentsFactory} to use.
@param vcrClusterAgentsFactory a {@link VcrClusterAgentsFactory} instance. Required if replicating from a VCR.
@param notificationSystem the {@link NotificationSystem} to use.
@param time The {@link Time} instance to use.
@param reporterFactory if non-null, use this function to set up a {@link JmxReporter} with custom settings. If this
                       option is null the default settings for the reporter will be used.
@throws InstantiationException if there was an error during startup.
ambry-server/src/main/java/com/github/ambry/server/AmbryServer.java