private void terminate(String reason, Exception cause, boolean clearMetrics, boolean join) {
    if (cause == null) {
      LOG.info("{} Closing source {} because: {}", logPeerId(), this.queueId, reason);
    } else {
      LOG.error(String.format("%s Closing source %s because an error occurred: %s", logPeerId(),
        this.queueId, reason), cause);
    }
    this.sourceRunning = false;
    if (initThread != null && Thread.currentThread() != initThread) {
      // This usually won't happen but anyway, let's wait until the initialization thread exits.
      // And notice that we may call terminate directly from the initThread so here we need to
      // avoid join on ourselves.
      initThread.interrupt();
      Threads.shutdown(initThread, this.sleepForRetries);
    }
    Collection<ReplicationSourceShipper> workers = workerThreads.values();

    for (ReplicationSourceShipper worker : workers) {
      worker.stopWorker();
      worker.entryReader.setReaderRunning(false);
    }

    if (this.replicationEndpoint != null) {
      this.replicationEndpoint.stop();
    }

    for (ReplicationSourceShipper worker : workers) {
      if (worker.isAlive() || worker.entryReader.isAlive()) {
        try {
          // Wait worker to stop
          Thread.sleep(this.sleepForRetries);
        } catch (InterruptedException e) {
          LOG.info("{} Interrupted while waiting {} to stop", logPeerId(), worker.getName());
          Thread.currentThread().interrupt();
        }
        // If worker still is alive after waiting, interrupt it
        if (worker.isAlive()) {
          worker.interrupt();
        }
        // If entry reader is alive after waiting, interrupt it
        if (worker.entryReader.isAlive()) {
          worker.entryReader.interrupt();
        }
      }
      if (!server.isAborted() && !server.isStopped()) {
        // If server is running and worker is already stopped but there was still entries batched,
        // we need to clear buffer used for non processed entries
        worker.clearWALEntryBatch();
      }
    }

    if (join) {
      for (ReplicationSourceShipper worker : workers) {
        Threads.shutdown(worker, this.sleepForRetries);
        LOG.info("{} ReplicationSourceWorker {} terminated", logPeerId(), worker.getName());
      }
      if (this.replicationEndpoint != null) {
        try {
          this.replicationEndpoint.awaitTerminated(sleepForRetries * maxRetriesMultiplier,
            TimeUnit.MILLISECONDS);
        } catch (TimeoutException te) {
          LOG.warn("{} Got exception while waiting for endpoint to shutdown "
            + "for replication source : {}", logPeerId(), this.queueId, te);
        }
      }
    }

    // Can be null in test context.
    if (this.metrics != null) {
      if (clearMetrics) {
        this.metrics.clear();
      } else {
        this.metrics.terminate();
      }
    }
  }
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