HBASE-27368 Do not need to throw IllegalStateException when peer is not active in ReplicationSource.initialize (#4779)

Signed-off-by: Xin Sun <ddupgs@gmail.com>
(cherry picked from commit 460bb47)

Author: Apache9

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java

@@ -425,7 +425,7 @@ WALEntryFilter getWalEntryFilter() {
     return walEntryFilter;
   }
 
-  protected final void uncaughtException(Thread t, Throwable e, ReplicationSourceManager manager,
+  private void uncaughtException(Thread t, Throwable e, ReplicationSourceManager manager,
     String peerId) {
     RSRpcServices.exitIfOOME(e);
     LOG.error("Unexpected exception in {} currentPath={}", t.getName(), getCurrentPath(), e);
@@ -552,15 +552,10 @@ private void initialize() {
     }
 
     if (!this.isSourceActive()) {
+      // this means the server is shutting down or the source is terminated, just give up
+      // initializing
       setSourceStartupStatus(false);
-      if (Thread.currentThread().isInterrupted()) {
-        // If source is not running and thread is interrupted this means someone has tried to
-        // remove this peer.
-        return;
-      }
-
-      retryStartup.set(!this.abortOnError);
-      throw new IllegalStateException("Source should be active.");
+      return;
     }
 
     sleepMultiplier = 1;
@@ -582,15 +577,12 @@ private void initialize() {
     }
 
     if (!this.isSourceActive()) {
+      // this means the server is shutting down or the source is terminated, just give up
+      // initializing
       setSourceStartupStatus(false);
-      if (Thread.currentThread().isInterrupted()) {
-        // If source is not running and thread is interrupted this means someone has tried to
-        // remove this peer.
-        return;
-      }
-      retryStartup.set(!this.abortOnError);
-      throw new IllegalStateException("Source should be active.");
+      return;
     }
+
     LOG.info("{} queueId={} (queues={}) is replicating from cluster={} to cluster={}", logPeerId(),
       this.replicationQueueInfo.getQueueId(), logQueue.getNumQueues(), clusterId, peerClusterId);
     initializeWALEntryFilter(peerClusterId);