private void initialize() {
    int sleepMultiplier = 1;
    while (this.isSourceActive()) {
      ReplicationEndpoint replicationEndpoint;
      try {
        replicationEndpoint = createReplicationEndpoint();
      } catch (Exception e) {
        LOG.warn("{} error creating ReplicationEndpoint, retry", logPeerId(), e);
        if (sleepForRetries("Error creating ReplicationEndpoint", sleepMultiplier)) {
          sleepMultiplier++;
        }
        continue;
      }

      try {
        initAndStartReplicationEndpoint(replicationEndpoint);
        this.replicationEndpoint = replicationEndpoint;
        break;
      } catch (Exception e) {
        LOG.warn("{} Error starting ReplicationEndpoint, retry", logPeerId(), e);
        replicationEndpoint.stop();
        if (sleepForRetries("Error starting ReplicationEndpoint", sleepMultiplier)) {
          sleepMultiplier++;
        } else {
          retryStartup.set(!this.abortOnError);
          setSourceStartupStatus(false);
          throw new RuntimeException("Exhausted retries to start replication endpoint.");
        }
      }
    }

    if (!this.isSourceActive()) {
      // this means the server is shutting down or the source is terminated, just give up
      // initializing
      setSourceStartupStatus(false);
      return;
    }

    sleepMultiplier = 1;
    UUID peerClusterId;
    // delay this until we are in an asynchronous thread
    for (;;) {
      peerClusterId = replicationEndpoint.getPeerUUID();
      if (this.isSourceActive() && peerClusterId == null) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("{} Could not connect to Peer ZK. Sleeping for {} millis", logPeerId(),
            (this.sleepForRetries * sleepMultiplier));
        }
        if (sleepForRetries("Cannot contact the peer's zk ensemble", sleepMultiplier)) {
          sleepMultiplier++;
        }
      } else {
        break;
      }
    }

    if (!this.isSourceActive()) {
      // this means the server is shutting down or the source is terminated, just give up
      // initializing
      setSourceStartupStatus(false);
      return;
    }

    LOG.info("{} queueId={} (queues={}) is replicating from cluster={} to cluster={}", logPeerId(),
      queueId, logQueue.getNumQueues(), clusterId, peerClusterId);
    initializeWALEntryFilter(peerClusterId);
    // Start workers
    startShippers();
    setSourceStartupStatus(false);
  }
Do the sleeping logic
@param msg             Why we sleep
@param sleepMultiplier by how many times the default sleeping time is augmented
@return True if <code>sleepMultiplier</code> is &lt; <code>maxRetriesMultiplier</code>
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java