docs: void startMigratePartitionLeader(boolean prepareStopNode) {
          RealVoltDB voltDB = (RealVoltDB)VoltDB.instance();
          final int hostId = CoreUtils.getHostIdFromHSId(m_siteId);
          Pair<Integer, Integer> target = null;
          if (prepareStopNode) {
              target = m_cartographer.getPartitionLeaderMigrationTargetForStopNode(hostId);
          } else {
              if (voltDB.isClusterComplete()) {
                  target = m_cartographer.getPartitionLeaderMigrationTarget(voltDB.getHostCount(), hostId, prepareStopNode);
              } else {
                  // Out of the scheduled task
                  target = new Pair<Integer, Integer> (-1, -1);
              }
          }
  
          //The host does not have any thing to do this time. It does not mean that the host does not
          //have more partition leaders than expected. Other hosts may have more partition leaders
          //than this one. So let other hosts do @MigratePartitionLeader first.
          if (target == null) {
              return;
          }
  
          final int partitionId = target.getFirst();
          final int targetHostId = target.getSecond();
          int partitionKey = -1;
  
          //MigratePartitionLeader is completed or there are hosts down. Stop MigratePartitionLeader service on this host
          if (targetHostId == -1 || (!prepareStopNode && !voltDB.isClusterComplete())) {
              voltDB.scheduleWork(
                      () -> {m_mailbox.deliver(new MigratePartitionLeaderMessage());},
                      0, 0, TimeUnit.SECONDS);
              return;
          }
  
          //Others may also iterate through the partition keys. So make a copy and find the key
          VoltTable partitionKeys = TheHashinator.getPartitionKeys(VoltType.INTEGER);
          ByteBuffer buf = ByteBuffer.allocate(partitionKeys.getSerializedSize());
          partitionKeys.flattenToBuffer(buf);
          buf.flip();
          VoltTable keyCopy = PrivateVoltTableFactory.createVoltTableFromSharedBuffer(buf);
          keyCopy.resetRowPosition();
          while (keyCopy.advanceRow()) {
              if (partitionId == keyCopy.getLong("PARTITION_ID")) {
                  partitionKey = (int)(keyCopy.getLong("PARTITION_KEY"));
                  break;
              }
          }
  
          if (partitionKey == -1) {
              tmLog.warn("Could not find the partition key for partition " + partitionId);
              return;
          }
  
          //grab a lock
          String errorMessage = VoltZK.createActionBlocker(m_zk, VoltZK.migratePartitionLeaderBlocker,
                  CreateMode.EPHEMERAL, tmLog,
                  "Migrate Partition Leader");
          if (errorMessage != null) {
              tmLog.rateLimitedLog(60, Level.INFO, null, errorMessage);
              return;
          }
  
          if (tmLog.isDebugEnabled()) {
              tmLog.debug(String.format("Move the leader of partition %d to host %d", partitionId, targetHostId));
              VoltTable vt = Cartographer.peekTopology(m_cartographer);
              tmLog.debug("[@MigratePartitionLeader]\n" + vt.toFormattedString());
          }
  
          boolean transactionStarted = false;
          Long targetHSId = m_cartographer.getHSIDForPartitionHost(targetHostId, partitionId);
          if (targetHSId == null) {
              if (tmLog.isDebugEnabled()) {
                  tmLog.debug(String.format("Partition %d is no longer on host %d", partitionId, targetHostId));
              }
              return;
          }
          try {
              SimpleClientResponseAdapter.SyncCallback cb = new SimpleClientResponseAdapter.SyncCallback();
              final String procedureName = "@MigratePartitionLeader";
              Config procedureConfig = SystemProcedureCatalog.listing.get(procedureName);
              StoredProcedureInvocation spi = new StoredProcedureInvocation();
              spi.setProcName(procedureName);
              spi.setClientHandle(m_executeTaskAdpater.registerCallback(cb));
              spi.setParams(partitionKey, partitionId, targetHostId);
              if (spi.getSerializedParams() == null) {
                  spi = MiscUtils.roundTripForCL(spi);
              }
  
              //Info saved for the node failure handling
              MigratePartitionLeaderInfo spiInfo = new MigratePartitionLeaderInfo(
                      m_cartographer.getHSIDForPartitionHost(hostId, partitionId),
                      targetHSId,
                      partitionId);
              VoltZK.createMigratePartitionLeaderInfo(m_zk, spiInfo);
  
              notifyPartitionMigrationStatus(partitionId, targetHSId, false);
  
              if (Boolean.getBoolean("TEST_MIGRATION_FAILURE")) {
                  Thread.sleep(100);
                  throw new IOException("failure simulation");
              }
              synchronized (m_executeTaskAdpater) {
                  if (createTransaction(m_executeTaskAdpater.connectionId(),
                          spi,
                          procedureConfig.getReadonly(),
                          procedureConfig.getSinglepartition(),
                          procedureConfig.getEverysite(),
                          partitionId,
                          spi.getSerializedSize(),
                          System.nanoTime()) != CreateTransactionResult.SUCCESS) {
                      tmLog.warn(String.format("Failed to start transaction for migration of partition %d to host %d",
                              partitionId, targetHostId));
                      notifyPartitionMigrationStatus(partitionId, targetHSId, true);
                      return;
                  }
              }
  
              transactionStarted = true;
  
              final long timeoutMS = 5 * 60 * 1000;
              ClientResponse resp = cb.getResponse(timeoutMS);
              if (resp != null && resp.getStatus() == ClientResponse.SUCCESS) {
                  tmLog.info(String.format("The partition leader for %d has been moved to host %d.",
                          partitionId, targetHostId));
              } else {
                  //not necessary a failure.
                  tmLog.warn(String.format("Fail to move the leader of partition %d to host %d. %s",
                          partitionId, targetHostId, resp == null ? null : resp.getStatusString()));
                  notifyPartitionMigrationStatus(partitionId, targetHSId, true);
              }
          } catch (Exception e) {
              tmLog.warn(String.format("errors in leader change for partition %d", partitionId), e);
              notifyPartitionMigrationStatus(partitionId, targetHSId, true);
          } finally {
              if (!transactionStarted) {
                  return;
              }
  
              //wait for the Cartographer to see the new partition leader. The leader promotion process should happen instantly.
              //If the new leader does not show up in 5 min, the cluster may have experienced host-down events.
              long remainingWaitTime = TimeUnit.MINUTES.toMillis(5);
              final long waitingInterval = TimeUnit.SECONDS.toMillis(1);
              boolean anyFailedHosts = false;
              boolean migrationComplete = false;
              while (remainingWaitTime > 0) {
                  try {
                      Thread.sleep(waitingInterval);
                  } catch (InterruptedException ignoreIt) {
                  }
                  remainingWaitTime -= waitingInterval;
                  if (CoreUtils.getHostIdFromHSId(m_cartographer.getHSIdForMaster(partitionId)) == targetHostId) {
                      migrationComplete = true;
                      break;
                  }
  
                  //some hosts may be down.
                  if (!voltDB.isClusterComplete() && !prepareStopNode) {
                      anyFailedHosts = true;
                      // If the target host is still alive, migration is still going on.
                      if (!voltDB.getHostMessenger().getLiveHostIds().contains(targetHostId)) {
                          break;
                      }
                  }
              }
  
              //if there are failed hosts, this blocker will be removed in RealVoltDB.handleHostsFailedForMigratePartitionLeader()
              if (!anyFailedHosts) {
                  voltDB.scheduleWork(
                          () -> removeMigrationZKNodes(),
                          5, 0, TimeUnit.SECONDS);
              }
  
              if (!migrationComplete) {
                  notifyPartitionMigrationStatus(partitionId, targetHSId, true);
              }
          }
      }