public boolean prepareServerForShutdown(DataNodeId dataNodeId, long logGrowthPauseLagThresholdBytes,
      short numReplicasCaughtUpPerPartition, long timeoutSecs, long checkRepeatDelaySecs)
      throws InterruptedException, IOException, TimeoutException {
    long startTimeSecs = time.seconds();
    boolean success = false;
    // 1. Make sure the peers of the node have caught up till the logGrowthPauseLagThresholdBytes
    LOGGER.info("Waiting until peers of {} are within {} bytes for all partitions", dataNodeId,
        logGrowthPauseLagThresholdBytes);
    if (waitTillCatchupOrTimeout(dataNodeId, logGrowthPauseLagThresholdBytes, numReplicasCaughtUpPerPartition,
        startTimeSecs + timeoutSecs, checkRepeatDelaySecs)) {
      // 2. Disable all requests that result in log growth (PUT, DELETE and inbound replication).
      LOGGER.info("Disabling PUT, DELETE and inbound replication on {} for all partitions", dataNodeId);
      if (controlLogGrowth(dataNodeId, false)) {
        // 3. Make sure the peers of the node have completely caught up
        LOGGER.info("Waiting until peers of {} have completely caught up for all partitions", dataNodeId);
        success = waitTillCatchupOrTimeout(dataNodeId, 0, numReplicasCaughtUpPerPartition, startTimeSecs + timeoutSecs,
            checkRepeatDelaySecs);
      }
    }
    LOGGER.info("Server safe for shutdown: {}", success);
    if (!success && !controlLogGrowth(dataNodeId, true)) {
      throw new IllegalStateException("Could not reset state of " + dataNodeId + ". Please reset manually");
    }
    return success;
  }
Prepares a server for safe shutdown by ensuring that all its peers have caught up to it. Does this by
1. Ensuring that all peers are within {@code logGrowthPauseLagThresholdBytes} for all partitions on
{@code dataNodeId}
2. Disabling PUT, DELETE and inbound replication on {@code dataNodeId}.
3. Ensuring that all peers are fully caught up (no lag) for all partitions on {@code dataNodeId}.
<p/>
If the operation fails after #2, an attempt is made to undo the changes in #2. If that fails, an
 {@link IllegalStateException} will be thrown.
@param dataNodeId the {@link DataNodeId} that has to be prepared for shutdown.
@param logGrowthPauseLagThresholdBytes the peer lag (in bytes) at which PUT, DELETE and inbound replication will
                                       be shut off on {@code dataNodeId}.
@param numReplicasCaughtUpPerPartition the number of replicas that have to be within {@code acceptableLagInBytes}
                                       (per partition). The min of this value or the total count of replicas - 1 is
                                       considered.
@param timeoutSecs the number of seconds after which the operation will be considered failed.
@param checkRepeatDelaySecs the number of seconds after which the catchup status will be checked again if the
                            previous check did not succeed.
@return {@code true} if the server is safe for shutdown (all peers are fully caught up). {@code false} otherwise.
@throws IllegalStateException if the operation failed and the reset of server state failed.
@throws InterruptedException if there are any interruptions while waiting for catch up.
@throws IOException if there is any I/O error.
@throws TimeoutException if there are timeout errors while making network requests.
ambry-tools/src/main/java/com/github/ambry/tools/admin/SafeServerShutdownTool.java