docs: @Override
      public void initialize(Configuration config) {
          hostLog.info("PID of this Volt process is " + CLibrary.getpid());
          ShutdownHooks.enableServerStopLogging();
          synchronized(m_startAndStopLock) {
              exitAfterMessage = false;
              // Handle multiple invocations of server thread in the same JVM.
              // by clearing static variables/properties which ModuleManager,
              // and Settings depend on
              ConfigFactory.clearProperty(Settings.CONFIG_DIR);
              ModuleManager.resetCacheRoot();
              CipherExecutor.SERVER.shutdown();
              CipherExecutor.CLIENT.shutdown();
  
              m_isRunningWithOldVerb = config.m_startAction.isLegacy();
  
              // check that this is a 64 bit VM
              if (System.getProperty("java.vm.name").contains("64") == false) {
                  hostLog.fatal("You are running on an unsupported (probably 32 bit) JVM. Exiting.");
                  System.exit(-1);
              }
  
              // print the ascii art!.
              // determine the edition
              // Check license availability
              // All above - not for init
              String edition = "Community Edition";
              if (config.m_startAction != StartAction.INITIALIZE) {
                  consoleLog.l7dlog( Level.INFO, LogKeys.host_VoltDB_StartupString.name(), null);
                  // load license API
                  if (config.m_pathToLicense == null) {
                      m_licenseApi = MiscUtils.licenseApiFactory();
                      if (m_licenseApi == null) {
                          hostLog.fatal("Unable to open license file in default directories");
                      }
                  } else {
                      m_licenseApi = MiscUtils.licenseApiFactory(config.m_pathToLicense);
                      if (m_licenseApi == null) {
                          hostLog.fatal("Unable to open license file in provided path: " + config.m_pathToLicense);
                      }
                  }
  
                  if (m_licenseApi == null) {
                      hostLog.fatal("Please contact sales@voltdb.com to request a license.");
                      VoltDB.crashLocalVoltDB(
                              "Failed to initialize license verifier. " + "See previous log message for details.", false,
                              null);
                  }
  
                  if (config.m_isEnterprise) {
                      if (m_licenseApi.isEnterprise()) {
                          edition = "Enterprise Edition";
                      }
                      if (m_licenseApi.isPro()) {
                          edition = "Pro Edition";
                      }
                      if (m_licenseApi.isEnterpriseTrial()) {
                          edition = "Enterprise Edition";
                      }
                      if (m_licenseApi.isProTrial()) {
                          edition = "Pro Edition";
                      }
                      if (m_licenseApi.isAWSMarketplace()) {
                          edition = "AWS Marketplace Edition";
                      }
                  }
  
                  // this also prints out the license type on the console
                  readBuildInfo(edition);
  
                  // print out the licensee on the license
                  if (config.m_isEnterprise) {
                      String licensee = m_licenseApi.licensee();
                      if ((licensee != null) && (licensee.length() > 0)) {
                          consoleLog.info(String.format("Licensed to: %s", licensee));
                      }
                  }
              }
  
              // Replay command line args that we can see
              StringBuilder sb = new StringBuilder(2048).append("Command line arguments: ");
              sb.append(System.getProperty("sun.java.command", "[not available]"));
              hostLog.info(sb.toString());
  
              List<String> iargs = ManagementFactory.getRuntimeMXBean().getInputArguments();
              sb.delete(0, sb.length()).append("Command line JVM arguments:");
              for (String iarg : iargs) {
                  sb.append(" ").append(iarg);
              }
              if (iargs.size() > 0) {
                  hostLog.info(sb.toString());
              } else {
                  hostLog.info("No JVM command line args known.");
              }
  
              sb.delete(0, sb.length()).append("Command line JVM classpath: ");
              sb.append(System.getProperty("java.class.path", "[not available]"));
              hostLog.info(sb.toString());
  
              if (config.m_startAction == StartAction.INITIALIZE) {
                  if (config.m_forceVoltdbCreate) {
                      deleteInitializationMarkers(config);
                  }
              }
  
              // If there's no deployment provide a default and put it under voltdbroot.
              if (config.m_pathToDeployment == null) {
                  try {
                      config.m_pathToDeployment = setupDefaultDeployment(hostLog, config.m_voltdbRoot);
                      config.m_deploymentDefault = true;
                  } catch (IOException e) {
                      VoltDB.crashLocalVoltDB("Failed to write default deployment.", false, null);
                      return;
                  }
              }
  
              ReadDeploymentResults readDepl = readPrimedDeployment(config);
  
              if (config.m_startAction == StartAction.INITIALIZE) {
                  if (config.m_forceVoltdbCreate && m_nodeSettings.clean()) {
                      String msg = "Archived previous snapshot directory to " + m_nodeSettings.getSnapshoth() + ".1";
                      consoleLog.info(msg);
                      hostLog.info(msg);
                  }
                  if (readDepl.deployment.getDr() != null && DrRoleType.XDCR.equals(readDepl.deployment.getDr().getRole())) {
                      // add default export configuration to DR conflict table
                      CatalogUtil.addExportConfigToDRConflictsTable(readDepl.deployment.getExport());
                  }
                  stageDeploymentFileForInitialize(config, readDepl.deployment);
                  stageSchemaFiles(config, readDepl.deployment.getDr() != null && DrRoleType.XDCR.equals(readDepl.deployment.getDr().getRole()));
                  stageInitializedMarker(config);
                  hostLog.info("Initialized VoltDB root directory " + config.m_voltdbRoot.getPath());
                  consoleLog.info("Initialized VoltDB root directory " + config.m_voltdbRoot.getPath());
                  VoltDB.exit(0);
              }
              if (config.m_startAction.isLegacy()) {
                  consoleLog.warn("The \"" + config.m_startAction.m_verb + "\" command is deprecated, please use \"init\" and \"start\" for your cluster operations.");
              }
  
              // config UUID is part of the status tracker.
              m_statusTracker = new NodeStateTracker();
              final File stagedCatalogLocation = new VoltFile(RealVoltDB.getStagedCatalogPath(config.m_voltdbRoot.getAbsolutePath()));
  
              if (config.m_startAction.isLegacy()) {
                  File rootFH = CatalogUtil.getVoltDbRoot(readDepl.deployment.getPaths());
                  File inzFH = new VoltFile(rootFH, VoltDB.INITIALIZED_MARKER);
                  if (inzFH.exists()) {
                      VoltDB.crashLocalVoltDB("Cannot use legacy start action "
                              + config.m_startAction + " on voltdbroot "
                              + rootFH + " that was initialized with the init command");
                      return;
                  }
                  //Case where you give primed deployment with -d look in ../../ for initialized marker.
                  //Also check if parents are config and voltdbroot
                  File cfile = (new File(config.m_pathToDeployment)).getParentFile();
                  if (cfile != null) {
                      rootFH = cfile.getParentFile();
                      if ("config".equals(cfile.getName()) && VoltDB.DBROOT.equals(rootFH.getName())) {
                          inzFH = new VoltFile(rootFH, VoltDB.INITIALIZED_MARKER);
                          if (inzFH.exists()) {
                              VoltDB.crashLocalVoltDB("Can not use legacy start action "
                                      + config.m_startAction + " on voltdbroot "
                                      + rootFH + " that was initialized with the init command");
                              return;
                          }
                      }
                  }
                  if (stagedCatalogLocation.isFile()) {
                      hostLog.warn("Initialized schema is present, but is being ignored and may be removed.");
                  }
              } else {
                  assert (config.m_startAction == StartAction.PROBE);
                  if (stagedCatalogLocation.isFile()) {
                      assert (config.m_pathToCatalog == null) : config.m_pathToCatalog;
                      config.m_pathToCatalog = stagedCatalogLocation.getAbsolutePath();
                  }
              }
  
              List<String> failed = m_nodeSettings.ensureDirectoriesExist();
              if (!failed.isEmpty()) {
                  String msg = "Unable to access or create the following directories:\n  - " +
                          Joiner.on("\n  - ").join(failed);
                  VoltDB.crashLocalVoltDB(msg);
                  return;
              }
  
              if (config.m_hostCount == VoltDB.UNDEFINED) {
                  config.m_hostCount = readDepl.deployment.getCluster().getHostcount();
              }
  
              // set the mode first thing
              m_mode = OperationMode.INITIALIZING;
              m_config = config;
              m_startMode = OperationMode.RUNNING;
  
              // set a bunch of things to null/empty/new for tests
              // which reuse the process
              m_safeMpTxnId = Long.MAX_VALUE;
              m_lastSeenMpTxnId = Long.MIN_VALUE;
              m_clientInterface = null;
              m_adminListener = null;
              m_commandLog = new DummyCommandLog();
              m_snmp = new DummySnmpTrapSender();
              m_messenger = null;
              m_opsRegistrar = new OpsRegistrar();
              m_snapshotCompletionMonitor = null;
              m_catalogContext = null;
              m_partitionCountStats = null;
              m_ioStats = null;
              m_memoryStats = null;
              m_commandLogStats = null;
              m_statsManager = null;
              m_restoreAgent = null;
              m_recoveryStartTime = System.currentTimeMillis();
              m_hostIdWithStartupCatalog = 0;
              m_pathToStartupCatalog = m_config.m_pathToCatalog;
              m_replicationActive = new AtomicBoolean(false);
              m_configLogger = null;
              ActivePlanRepository.clear();
  
              updateMaxThreadsLimit();
  
              // set up site structure
              final int computationThreads = Math.max(2, CoreUtils.availableProcessors() / 4);
              m_computationService =
                      CoreUtils.getListeningExecutorService(
                              "Computation service thread",
                              computationThreads, m_config.m_computationCoreBindings);
  
              // Set std-out/err to use the UTF-8 encoding and fail if UTF-8 isn't supported
              try {
                  System.setOut(new PrintStream(System.out, true, "UTF-8"));
                  System.setErr(new PrintStream(System.err, true, "UTF-8"));
              } catch (UnsupportedEncodingException e) {
                  hostLog.fatal("Support for the UTF-8 encoding is required for VoltDB. This means you are likely running an unsupported JVM. Exiting.");
                  VoltDB.exit(-1);
              }
  
              m_snapshotCompletionMonitor = new SnapshotCompletionMonitor();
  
              // use CLI overrides for testing hotfix version compatibility
              if (m_config.m_versionStringOverrideForTest != null) {
                  m_versionString = m_config.m_versionStringOverrideForTest;
              }
              if (m_config.m_versionCompatibilityRegexOverrideForTest != null) {
                  m_hotfixableRegexPattern = m_config.m_versionCompatibilityRegexOverrideForTest;
              }
              if (m_config.m_buildStringOverrideForTest != null) {
                  m_buildString = m_config.m_buildStringOverrideForTest;
              }
              // Prime cluster settings from configuration parameters
              // evaluate properties with the following sources in terms of priority
              // 1) properties from command line options
              // 2) properties from the cluster.properties files
              // 3) properties from the deployment file
  
              // this reads the file config/cluster.properties
              ClusterSettings fromPropertyFile = ClusterSettings.create();
              // handle case we recover clusters that were elastically expanded
              if (m_config.m_startAction.doesRecover()) {
                  m_config.m_hostCount = fromPropertyFile.hostcount();
              }
              if (m_config.m_restorePlacement && !StringUtils.isEmpty(fromPropertyFile.partitionIds())) {
                  m_config.m_recoveredPartitions = fromPropertyFile.partitionIds();
              }
  
              Map<String, String> fromCommandLine = m_config.asClusterSettingsMap();
              Map<String, String> fromDeploymentFile = CatalogUtil.
                      asClusterSettingsMap(readDepl.deployment);
  
              ClusterSettings clusterSettings = ClusterSettings.create(
                      fromCommandLine, fromPropertyFile.asMap(), fromDeploymentFile);
  
              clusterSettings.store();
              m_clusterSettings.set(clusterSettings, 1);
  
              MeshProber.Determination determination = buildClusterMesh(readDepl);
              if (m_config.m_startAction == StartAction.PROBE) {
                  String action = "Starting a new database cluster";
                  if (determination.startAction.doesRejoin()) {
                      action = "Rejoining a running cluster";
                  } else if (determination.startAction == StartAction.JOIN) {
                      action = "Adding this node to a running cluster";
                  } else if (determination.startAction.doesRecover()) {
                      action = "Restarting the database cluster from the command logs";
                  }
                  hostLog.info(action);
                  consoleLog.info(action);
              }
  
              m_config.m_startAction = determination.startAction;
              m_config.m_hostCount = determination.hostCount;
              assert determination.paused || !m_config.m_isPaused;
              m_config.m_isPaused = determination.paused;
              m_terminusNonce = determination.terminusNonce;
  
              // determine if this is a rejoining node
              // (used for license check and later the actual rejoin)
              m_rejoining = m_config.m_startAction.doesRejoin();
              m_rejoinDataPending = m_config.m_startAction.doesJoin();
              m_meshDeterminationLatch.countDown();
              m_joining = m_config.m_startAction == StartAction.JOIN;
  
              if (m_rejoining || m_joining) {
                  m_statusTracker.set(NodeState.REJOINING);
              }
              //Register dummy agents immediately
              m_opsRegistrar.registerMailboxes(m_messenger);
  
              //Start validating the build string in the background
              final Future<?> buildStringValidation = validateBuildString(getBuildString(), m_messenger.getZK());
  
              // race to create start action nodes and then verify theirs compatibility.
              m_messenger.getZK().create(VoltZK.start_action, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new ZKUtil.StringCallback(), null);
              VoltZK.createStartActionNode(m_messenger.getZK(), m_messenger.getHostId(), m_config.m_startAction);
              validateStartAction();
  
              if (m_rejoining) {
                  //grab rejoining lock before catalog read
                  Iv2RejoinCoordinator.acquireLock(m_messenger);
              }
  
              m_durable = readDeploymentAndCreateStarterCatalogContext(config);
  
              if (config.m_isEnterprise && m_config.m_startAction.doesRequireEmptyDirectories()
                      && !config.m_forceVoltdbCreate && m_durable) {
                  managedPathsEmptyCheck(config);
              }
  
              // wait to make sure every host actually *see* each other's ZK node state.
              final int numberOfNodes = m_messenger.getLiveHostIds().size();
              Map<Integer, HostInfo> hostInfos = m_messenger.waitForGroupJoin(numberOfNodes);
              if (m_messenger.isPaused() || m_config.m_isPaused) {
                  setStartMode(OperationMode.PAUSED);
              }
  
              // Create the thread pool here.
              m_periodicWorkThread =
                      CoreUtils.getScheduledThreadPoolExecutor("Periodic Work", 1, CoreUtils.SMALL_STACK_SIZE);
              m_periodicPriorityWorkThread =
                      CoreUtils.getScheduledThreadPoolExecutor("Periodic Priority Work", 1, CoreUtils.SMALL_STACK_SIZE);
  
              m_snapshotIOAgent = new SnapshotIOAgentImpl(m_messenger,
                      m_messenger.getHSIdForLocalSite(HostMessenger.SNAPSHOT_IO_AGENT_ID));
              m_messenger.createMailbox(m_snapshotIOAgent.getHSId(), m_snapshotIOAgent);
  
              try {
                  SimpleDateFormat sdf = new SimpleDateFormat("EEE MMM d, yyyy");
                  JSONObject jo = new JSONObject();
                  jo.put("trial", m_licenseApi.isAnyKindOfTrial());
                  jo.put("hostcount",m_licenseApi.maxHostcount());
                  jo.put("commandlogging", m_licenseApi.isCommandLoggingAllowed());
                  jo.put("wanreplication", m_licenseApi.isDrReplicationAllowed());
                  jo.put("expiration", sdf.format(m_licenseApi.expires().getTime()));
                  m_licenseInformation = jo.toString();
              } catch (JSONException ex) {
                  //Ignore
              }
  
              // Create the GlobalServiceElector.  Do this here so we can register the MPI with it
              // when we construct it below
              m_globalServiceElector = new GlobalServiceElector(m_messenger.getZK(), m_messenger.getHostId());
  
              // Always create a mailbox for elastic service data transfer
              if (m_config.m_isEnterprise) {
                  long elasticHSId = m_messenger.getHSIdForLocalSite(HostMessenger.REBALANCE_SITE_ID);
                  m_messenger.createMailbox(elasticHSId, new SiteMailbox(m_messenger, elasticHSId));
              }
  
              if (m_joining) {
                  try {
                      int kfactor = m_catalogContext.getDeployment().getCluster().getKfactor();
                       m_joinCoordinator = ProClass.newInstanceOf("org.voltdb.elastic.ElasticJoinNodeCoordinator", "Elastic",
                              ProClass.HANDLER_LOG, m_messenger, VoltDB.instance().getVoltDBRootPath(), kfactor);
                      if (!MiscUtils.validateLicense(getLicenseApi(),
                              m_clusterSettings.get().hostcount() + m_joinCoordinator.getHostsJoining(),
                              DrRoleType.fromValue(getCatalogContext().getCluster().getDrrole()),
                              m_config.m_startAction)) {
                          VoltDB.crashLocalVoltDB("VoltDB license constraints are not met.");
                      }
                      if (determination.startAction == StartAction.JOIN && m_joinCoordinator.getHostsJoining() > 1) {
                          String waitMessage = "The join process will begin after a total of "
                                  + m_joinCoordinator.getHostsJoining() + " nodes are added, waiting...";
                          consoleLog.info(waitMessage);
                      }
                      m_messenger.registerMailbox(m_joinCoordinator);
                      m_joinCoordinator.initialize();
                  } catch (Exception e) {
                      VoltDB.crashLocalVoltDB("Failed to instantiate join coordinator", true, e);
                  }
              }
  
              /*
               * Construct all the mailboxes for things that need to be globally addressable so they can be published
               * in one atomic shot.
               *
               * The starting state for partition assignments are statically derived from the host id generated
               * by host messenger and the k-factor/host count/sites per host. This starting state
               * is published to ZK as the topology metadata node.
               *
               * On join and rejoin the node has to inspect the topology meta node to find out what is missing
               * and then update the topology listing itself as the replica for those partitions.
               * Then it does a compare and set of the topology.
               *
               * Ning: topology may not reflect the true partitions in the cluster during join. So if another node
               * is trying to rejoin, it should rely on the cartographer's view to pick the partitions to replace.
               */
              AbstractTopology topo = getTopology(config.m_startAction, hostInfos, m_joinCoordinator);
              m_partitionsToSitesAtStartupForExportInit = new ArrayList<>();
              try {
                  // IV2 mailbox stuff
                  m_cartographer = new Cartographer(m_messenger, m_configuredReplicationFactor,
                          m_catalogContext.cluster.getNetworkpartition());
                  m_partitionZeroLeader = new Supplier<Boolean>() {
                      @Override
                      public Boolean get() {
                          return m_cartographer.isPartitionZeroLeader();
                      }
                  };
                  List<Integer> partitions = null;
                  Set<Integer> partitionGroupPeers = null;
                  if (m_rejoining) {
                      m_configuredNumberOfPartitions = m_cartographer.getPartitionCount();
                      Set<Integer> recoverPartitions = null;
                      if (m_config.m_restorePlacement) {
                          recoverPartitions = hostInfos.get(m_messenger.getHostId()).getRecoveredPartitions();
                      }
                      AbstractTopology recoveredTopo = recoverPartitions(topo, hostInfos.get(m_messenger.getHostId()).m_group, recoverPartitions);
                      if (recoveredTopo != null) {
                          topo = recoveredTopo;
                          partitions = Lists.newArrayList(topo.getPartitionIdList(m_messenger.getHostId()));
                      }
                      if (partitions == null) {
                          partitions = m_cartographer.getIv2PartitionsToReplace(m_configuredReplicationFactor,
                                  m_catalogContext.getNodeSettings().getLocalSitesCount(), m_messenger.getHostId(),
                                  Maps.transformValues(hostInfos, h -> h.m_group));
                      }
                      partitionGroupPeers = m_cartographer.findPartitionGroupPeers(partitions);
                      if (partitions.size() == 0) {
                          VoltDB.crashLocalVoltDB("The VoltDB cluster already has enough nodes to satisfy " +
                                  "the requested k-safety factor of " +
                                  m_configuredReplicationFactor + ".\n" +
                                  "No more nodes can join.", false, null);
                      }
                  } else {
                      m_configuredNumberOfPartitions = topo.getPartitionCount();
                      partitions = Lists.newArrayList(topo.getPartitionIdList(m_messenger.getHostId()));
                      partitionGroupPeers = topo.getPartitionGroupPeers(m_messenger.getHostId());
                  }
  
                  m_eligibleAsLeader = determineIfEligibleAsLeader(partitions, partitionGroupPeers, topo);
  
                  m_messenger.setPartitionGroupPeers(partitionGroupPeers, m_clusterSettings.get().hostcount());
  
                  // The partition id list must be in sorted order
                  assert(Ordering.natural().isOrdered(partitions));
  
                  // persist the merged settings
                  m_config.m_recoveredPartitions = Joiner.on(",").join(partitions);
                  clusterSettings = ClusterSettings.create(
                          m_config.asClusterSettingsMap(), fromPropertyFile.asMap(), fromDeploymentFile);
                  clusterSettings.store();
                  hostLog.info("Partitions on this host:" + m_config.m_recoveredPartitions);
                  for (Integer partition : partitions) {
                      m_iv2InitiatorStartingTxnIds.put(partition, TxnEgo.makeZero(partition).getTxnId());
                  }
                  m_iv2Initiators = createIv2Initiators(
                          partitions,
                          m_config.m_startAction,
                          m_partitionsToSitesAtStartupForExportInit);
                  m_iv2InitiatorStartingTxnIds.put(MpInitiator.MP_INIT_PID,
                          TxnEgo.makeZero(MpInitiator.MP_INIT_PID).getTxnId());
  
                  if (m_eligibleAsLeader) {
                      // Start the GlobalServiceElector. Not sure where this will actually belong.
                      try {
                          m_globalServiceElector.start();
                      } catch (Exception e) {
                          VoltDB.crashLocalVoltDB("Unable to start GlobalServiceElector", true, e);
                      }
  
                      // Pass the local HSIds to the MPI so it can farm out buddy sites
                      // to the RO MP site pool
                      List<Long> localHSIds = new ArrayList<>();
                      for (Initiator ii : m_iv2Initiators.values()) {
                          localHSIds.add(ii.getInitiatorHSId());
                      }
  
                      m_MPI = new MpInitiator(m_messenger, localHSIds, getStatsAgent());
                      m_iv2Initiators.put(MpInitiator.MP_INIT_PID, m_MPI);
                  }
  
                  // Make a list of HDIds to join
                  Map<Integer, Long> partsToHSIdsToRejoin = new HashMap<>();
                  for (Initiator init : m_iv2Initiators.values()) {
                      if (init.isRejoinable()) {
                          partsToHSIdsToRejoin.put(init.getPartitionId(), init.getInitiatorHSId());
                      }
                  }
                  OnDemandBinaryLogger.path = VoltDB.instance().getVoltDBRootPath();
                  if (m_rejoining) {
                      SnapshotSaveAPI.recoveringSiteCount.set(partsToHSIdsToRejoin.size());
                      hostLog.info("Set recovering site count to " + partsToHSIdsToRejoin.size());
  
                      m_joinCoordinator = new Iv2RejoinCoordinator(m_messenger,
                              partsToHSIdsToRejoin.values(),
                              VoltDB.instance().getVoltDBRootPath(),
                              m_config.m_startAction == StartAction.LIVE_REJOIN);
                      m_joinCoordinator.initialize();
                      m_messenger.registerMailbox(m_joinCoordinator);
                      if (m_config.m_startAction == StartAction.LIVE_REJOIN) {
                          hostLog.info("Using live rejoin.");
                      }
                      else {
                          hostLog.info("Using blocking rejoin.");
                      }
                  } else if (m_joining) {
                      m_joinCoordinator.setPartitionsToHSIds(partsToHSIdsToRejoin);
                  }
              } catch (Exception e) {
                  VoltDB.crashLocalVoltDB(e.getMessage(), true, e);
              }
  
              // do the many init tasks in the Inits class
              Inits inits = new Inits(m_statusTracker, this, 1, m_durable);
              inits.doInitializationWork();
  
              int kfactor = m_catalogContext.getDeployment().getCluster().getKfactor();
              if(determination.startAction == StartAction.JOIN && kfactor > 0) {
                  int kfactorPlusOne = kfactor + 1;
                  String waitMessage = "" + kfactorPlusOne + " nodes added, joining new nodes to the cluster...";
                  consoleLog.info(waitMessage);
              }
  
              // Need the catalog so that we know how many tables so we can guess at the necessary heap size
              // This is done under Inits.doInitializationWork(), so need to wait until we get here.
              // Current calculation needs pro/community knowledge, number of tables, and the sites/host,
              // which is the number of initiators (minus the possibly idle MPI initiator)
              checkHeapSanity(m_catalogContext.tables.size(),
                      (getLocalPartitionCount()), m_configuredReplicationFactor);
  
              if (m_joining) {
                  if (hostLog.isDebugEnabled()) {
                      hostLog.debug("Elastic join happened on a cluster of DR Role:" + getReplicationRole());
                  }
              }
  
              collectLocalNetworkMetadata();
  
              // Initialize stats
              m_ioStats = new IOStats();
              getStatsAgent().registerStatsSource(StatsSelector.IOSTATS,
                      0, m_ioStats);
              m_memoryStats = new MemoryStats();
              getStatsAgent().registerStatsSource(StatsSelector.MEMORY,
                      0, m_memoryStats);
              getStatsAgent().registerStatsSource(StatsSelector.TOPO, 0, m_cartographer);
              m_partitionCountStats = new PartitionCountStats(m_cartographer);
              getStatsAgent().registerStatsSource(StatsSelector.PARTITIONCOUNT,
                      0, m_partitionCountStats);
              m_initiatorStats = new InitiatorStats(m_myHostId);
              m_liveClientsStats = new LiveClientsStats();
              getStatsAgent().registerStatsSource(StatsSelector.LIVECLIENTS, 0, m_liveClientsStats);
  
              m_latencyStats = new LatencyStats();
              getStatsAgent().registerStatsSource(StatsSelector.LATENCY, 0, m_latencyStats);
              m_latencyCompressedStats = new LatencyHistogramStats(m_myHostId);
              getStatsAgent().registerStatsSource(StatsSelector.LATENCY_COMPRESSED, 0, m_latencyCompressedStats);
              m_latencyHistogramStats = new LatencyUncompressedHistogramStats(m_myHostId);
              getStatsAgent().registerStatsSource(StatsSelector.LATENCY_HISTOGRAM,
                      0, m_latencyHistogramStats);
  
  
              BalancePartitionsStatistics rebalanceStats = new BalancePartitionsStatistics();
              getStatsAgent().registerStatsSource(StatsSelector.REBALANCE, 0, rebalanceStats);
  
              KSafetyStats kSafetyStats = new KSafetyStats();
              getStatsAgent().registerStatsSource(StatsSelector.KSAFETY, 0, kSafetyStats);
              m_cpuStats = new CpuStats();
              getStatsAgent().registerStatsSource(StatsSelector.CPU,
                      0, m_cpuStats);
              m_gcStats = new GcStats();
              getStatsAgent().registerStatsSource(StatsSelector.GC,
                      0, m_gcStats);
              // ENG-6321
              m_commandLogStats = new CommandLogStats(m_commandLog);
              getStatsAgent().registerStatsSource(StatsSelector.COMMANDLOG, 0, m_commandLogStats);
  
              // Dummy DRCONSUMER stats
              replaceDRConsumerStatsWithDummy();
  
              /*
               * Initialize the command log on rejoin and join before configuring the IV2
               * initiators.  This will prevent them from receiving transactions
               * which need logging before the internal file writers are
               * initialized.  Root cause of ENG-4136.
               *
               * If sync command log is on, not initializing the command log before the initiators
               * are up would cause deadlock.
               */
              if ((m_commandLog != null) && (m_commandLog.needsInitialization())) {
                  consoleLog.l7dlog(Level.INFO, LogKeys.host_VoltDB_StayTunedForLogging.name(), null);
              }
              else {
                  consoleLog.l7dlog(Level.INFO, LogKeys.host_VoltDB_StayTunedForNoLogging.name(), null);
              }
              if (m_commandLog != null && (m_rejoining || m_joining)) {
                  //On rejoin the starting IDs are all 0 so technically it will load any snapshot
                  //but the newest snapshot will always be the truncation snapshot taken after rejoin
                  //completes at which point the node will mark itself as actually recovered.
                  //
                  // Use the partition count from the cluster config instead of the cartographer
                  // here. Since the initiators are not started yet, the cartographer still doesn't
                  // know about the new partitions at this point.
                  m_commandLog.initForRejoin(
                          m_catalogContext.cluster.getLogconfig().get("log").getLogsize(),
                          Long.MIN_VALUE,
                          true,
                          m_config.m_commandLogBinding, m_iv2InitiatorStartingTxnIds);
              }
  
              // Create the client interface
              try {
                  InetAddress clientIntf = null;
                  InetAddress adminIntf = null;
                  if (!m_config.m_externalInterface.trim().equals("")) {
                      clientIntf = InetAddress.getByName(m_config.m_externalInterface);
                      //client and admin interfaces are same by default.
                      adminIntf = clientIntf;
                  }
                  //If user has specified on command line host:port override client and admin interfaces.
                  if (m_config.m_clientInterface != null && m_config.m_clientInterface.trim().length() > 0) {
                      clientIntf = InetAddress.getByName(m_config.m_clientInterface);
                  }
                  if (m_config.m_adminInterface != null && m_config.m_adminInterface.trim().length() > 0) {
                      adminIntf = InetAddress.getByName(m_config.m_adminInterface);
                  }
                  m_clientInterface = ClientInterface.create(m_messenger, m_catalogContext, getReplicationRole(),
                          m_cartographer,
                          clientIntf,
                          config.m_port,
                          adminIntf,
                          config.m_adminPort,
                          m_config.m_sslExternal ? m_config.m_sslServerContext : null);
              } catch (Exception e) {
                  VoltDB.crashLocalVoltDB(e.getMessage(), true, e);
              }
  
              // DR overflow directory
              if (VoltDB.instance().getLicenseApi().isDrReplicationAllowed()) {
                  m_producerDRGateway = ProClass.newInstanceOf("org.voltdb.dr2.DRProducer", "DR Producer",
                          ProClass.HANDLER_CRASH,
                          new VoltFile(VoltDB.instance().getDROverflowPath()),
                          new VoltFile(VoltDB.instance().getSnapshotPath()),
                          willDoActualRecover() ? ProducerDRGateway.Mode.RECOVER
                                  : m_config.m_startAction.doesRejoin() ? ProducerDRGateway.Mode.REJOIN
                                          : m_config.m_startAction == StartAction.JOIN ? ProducerDRGateway.Mode.JOIN
                                                  : ProducerDRGateway.Mode.NEW,
                          m_replicationActive.get(), m_configuredNumberOfPartitions,
                          (m_catalogContext.getClusterSettings().hostcount() - m_config.m_missingHostCount));
              }
              else {
                  // set up empty stats for the DR Producer
                  getStatsAgent().registerStatsSource(StatsSelector.DRPRODUCERNODE, 0,
                          new DRProducerStatsBase.DRProducerNodeStatsBase());
                  getStatsAgent().registerStatsSource(StatsSelector.DRPRODUCERPARTITION, 0,
                          new DRProducerStatsBase.DRProducerPartitionStatsBase());
              }
              m_drRoleStats = new DRRoleStats(this);
              getStatsAgent().registerStatsSource(StatsSelector.DRROLE, 0, m_drRoleStats);
  
              /*
               * Configure and start all the IV2 sites
               */
              try {
                  final String serializedCatalog = m_catalogContext.catalog.serialize();
                  for (Initiator iv2init : m_iv2Initiators.values()) {
                      iv2init.configure(
                              getBackendTargetType(),
                              m_catalogContext,
                              serializedCatalog,
                              m_configuredNumberOfPartitions,
                              m_config.m_startAction,
                              getStatsAgent(),
                              m_memoryStats,
                              m_commandLog,
                              m_config.m_executionCoreBindings.poll(),
                              isLowestSiteId(iv2init));
                  }
  
                  // LeaderAppointer startup blocks if the initiators are not initialized.
                  // So create the LeaderAppointer after the initiators.
                  boolean expectSyncSnapshot = getReplicationRole() == ReplicationRole.REPLICA && config.m_startAction == StartAction.CREATE;
                  m_leaderAppointer = new LeaderAppointer(
                          m_messenger,
                          m_configuredNumberOfPartitions,
                          m_catalogContext.getDeployment().getCluster().getKfactor(),
                          topo,
                          m_MPI,
                          kSafetyStats,
                          expectSyncSnapshot
                  );
                  m_globalServiceElector.registerService(m_leaderAppointer);
              } catch (Exception e) {
                  Throwable toLog = e;
                  if (e instanceof ExecutionException) {
                      toLog = ((ExecutionException)e).getCause();
                  }
                  VoltDB.crashLocalVoltDB("Error configuring IV2 initiator.", true, toLog);
              }
  
              // Create the statistics manager and register it to JMX registry
              m_statsManager = null;
              try {
                  if (!DISABLE_JMX) {
                      m_statsManager = ProClass.newInstanceOf("org.voltdb.management.JMXStatsManager", "JMX",
                              ProClass.HANDLER_IGNORE);
                      if (m_statsManager != null) {
                          m_statsManager.initialize();
                      }
                  }
              } catch (Exception e) {
                  //JMXStatsManager will log and we continue.
              }
  
              try {
                  m_snapshotCompletionMonitor.init(m_messenger.getZK());
              } catch (Exception e) {
                  hostLog.fatal("Error initializing snapshot completion monitor", e);
                  VoltDB.crashLocalVoltDB("Error initializing snapshot completion monitor", true, e);
              }
  
              /*
               * Make sure the build string successfully validated
               * before continuing to do operations
               * that might return wrongs answers or lose data.
               */
              try {
                  buildStringValidation.get();
              } catch (Exception e) {
                  VoltDB.crashLocalVoltDB("Failed to validate cluster build string", false, e);
              }
  
              //elastic join, make sure all the joining nodes are ready
              //so that the secondary connections can be created.
              if (m_joining) {
                  int expectedHosts = m_configuredReplicationFactor + 1;
                  m_messenger.waitForJoiningHostsToBeReady(expectedHosts, this.m_myHostId);
              } else if (!m_rejoining) {
                  // initial start or recover
                  int expectedHosts = m_catalogContext.getClusterSettings().hostcount() - m_config.m_missingHostCount;
                  m_messenger.waitForAllHostsToBeReady(expectedHosts);
              }
  
              // @hostFailed() may not be triggered if there are host failures during mesh determination
              // Fail this rejoining node here if it sees site failure
              if (m_messenger.getFailedSiteCount() > 0) {
                  stopRejoiningHost();
              }
  
              //The connections between peers in partition groups could be slow
              //The problem will be addressed.
              if (!(m_config.m_sslInternal)) {
                  // Create secondary connections within partition group
                  createSecondaryConnections(m_rejoining);
              }
  
              if (!m_joining && (m_cartographer.getPartitionCount()) != m_configuredNumberOfPartitions) {
                  for (Map.Entry<Integer, ImmutableList<Long>> entry :
                      getSiteTrackerForSnapshot().m_partitionsToSitesImmutable.entrySet()) {
                      hostLog.info(entry.getKey() + " -- "
                              + CoreUtils.hsIdCollectionToString(entry.getValue()));
                  }
                  VoltDB.crashGlobalVoltDB("Mismatch between configured number of partitions (" +
                          m_configuredNumberOfPartitions + ") and actual (" +
                          m_cartographer.getPartitionCount() + ")",
                          true, null);
              }
  
              schedulePeriodicWorks();
              m_clientInterface.schedulePeriodicWorks();
  
              // print out a bunch of useful system info
              logDebuggingInfo(m_config, m_httpPortExtraLogMessage, m_jsonEnabled);
  
  
              // warn the user on the console if k=0 or if no command logging
              if (m_configuredReplicationFactor == 0) {
                  consoleLog.warn("This is not a highly available cluster. K-Safety is set to 0.");
              }
              boolean usingCommandLog = m_config.m_isEnterprise
                      && (m_catalogContext.cluster.getLogconfig() != null)
                      && (m_catalogContext.cluster.getLogconfig().get("log") != null)
                      && m_catalogContext.cluster.getLogconfig().get("log").getEnabled();
              if (!usingCommandLog) {
                  // figure out if using a snapshot schedule
                  boolean usingPeridoicSnapshots = false;
                  for (SnapshotSchedule ss : m_catalogContext.database.getSnapshotschedule()) {
                      if (ss.getEnabled()) {
                          usingPeridoicSnapshots = true;
                      }
                  }
                  // print the right warning depending on durability settings
                  if (usingPeridoicSnapshots) {
                      consoleLog.warn("Durability is limited to periodic snapshots. Command logging is off.");
                  }
                  else {
                      consoleLog.warn("Durability is turned off. Command logging is off.");
                  }
              }
  
              // warn if cluster is partitionable, but partition detection is off
              if ((m_catalogContext.cluster.getNetworkpartition() == false) &&
                      (m_configuredReplicationFactor > 0)) {
                  hostLog.warn("Running a redundant (k-safe) cluster with network " +
                          "partition detection disabled is not recommended for production use.");
                  // we decided not to include the stronger language below for the 3.0 version (ENG-4215)
                  //hostLog.warn("With partition detection disabled, data may be lost or " +
                  //      "corrupted by certain classes of network failures.");
              }
  
              assert (m_clientInterface != null);
              m_clientInterface.initializeSnapshotDaemon(m_messenger, m_globalServiceElector);
              TTLManager.initialze();
              getStatsAgent().registerStatsSource(StatsSelector.TTL, 0, TTLManager.instance());
              // Start elastic services
              try {
                  if (m_config.m_isEnterprise) {
                      m_elasticService = ProClass.newInstanceOf("org.voltdb.elastic.ElasticCoordinator",
                              "Elastic Operations", ProClass.HANDLER_CRASH, m_messenger, m_clientInterface,
                              m_cartographer, rebalanceStats, VoltDB.instance().getCommandLogSnapshotPath(),
                              m_catalogContext.getDeployment().getCluster().getKfactor(), m_clusterSettings,
                              m_eligibleAsLeader);
                      m_elasticService.updateConfig(m_catalogContext);
                  }
              } catch (Exception e) {
                  VoltDB.crashLocalVoltDB("Failed to instantiate elastic services", false, e);
              }
  
              // set additional restore agent stuff
              if (m_restoreAgent != null) {
                  m_restoreAgent.setInitiator(new Iv2TransactionCreator(m_clientInterface));
              }
  
              // Start the stats agent at the end, after everything has been constructed
              m_opsRegistrar.setDummyMode(false);
  
              m_configLogger = new Thread(new ConfigLogging());
              m_configLogger.start();
  
              scheduleDailyLoggingWorkInNextCheckTime();
  
              // Write a message to the log file if LARGE_MODE_RATIO system property is set to anything other than 0.
              // This way it will be possible to verify that various frameworks are exercising large mode.
  
              // If -Dlarge_mode_ratio=xx is specified via ant, the value will show up in the environment variables and
              // take higher priority. Otherwise, the value specified via VOLTDB_OPTS will take effect.
              // If the test is started by ant and -Dlarge_mode_ratio is not set, it will take a default value "-1" which
              // we should ignore.
              final double largeModeRatio = Double.valueOf((System.getenv("LARGE_MODE_RATIO") == null ||
                      System.getenv("LARGE_MODE_RATIO").equals("-1")) ? System.getProperty("LARGE_MODE_RATIO", "0") : System.getenv("LARGE_MODE_RATIO"));
              if (largeModeRatio > 0) {
                  hostLog.info(String.format("The large_mode_ratio property is set as %.2f", largeModeRatio));
              }
          }
      }