HBASE-28129 Do not retry refreshSources when region server is already stopping

Author: Apache9

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

@@ -373,8 +373,8 @@ private void tryStartNewShipper(String walGroupId) {
         Threads.setDaemonThreadRunning(
           walReader, Thread.currentThread().getName() + ".replicationSource.wal-reader."
             + walGroupId + "," + queueId,
-          (t, e) -> this.uncaughtException(t, e, this.manager, this.getPeerId()));
-        worker.startup((t, e) -> this.uncaughtException(t, e, this.manager, this.getPeerId()));
+          (t, e) -> this.retryRefreshing(t, e, this.manager, this.getPeerId()));
+        worker.startup((t, e) -> this.retryRefreshing(t, e, this.manager, this.getPeerId()));
         return worker;
       }
     });
@@ -448,22 +448,26 @@ WALEntryFilter getWalEntryFilter() {
     return walEntryFilter;
   }
 
-  private void uncaughtException(Thread t, Throwable e, ReplicationSourceManager manager,
+  private void retryRefreshing(Thread t, Throwable error, ReplicationSourceManager manager,
     String peerId) {
-    OOMEChecker.exitIfOOME(e, getClass().getSimpleName());
-    LOG.error("Unexpected exception in {} currentPath={}", t.getName(), getCurrentPath(), e);
+    OOMEChecker.exitIfOOME(error, getClass().getSimpleName());
+    LOG.error("Unexpected exception in {} currentPath={}", t.getName(), getCurrentPath(), error);
     if (abortOnError) {
-      server.abort("Unexpected exception in " + t.getName(), e);
+      server.abort("Unexpected exception in " + t.getName(), error);
     }
     if (manager != null) {
       while (true) {
+        if (server.isAborted() || server.isStopped() || server.isStopping()) {
+          LOG.warn("Server is shutting down, give up refreshing source for peer {}", peerId);
+          return;
+        }
         try {
           LOG.info("Refreshing replication sources now due to previous error on thread: {}",
             t.getName());
           manager.refreshSources(peerId);
           break;
-        } catch (IOException | ReplicationException e1) {
-          LOG.error("Replication sources refresh failed.", e1);
+        } catch (Exception e) {
+          LOG.error("Replication sources refresh failed.", e);
           sleepForRetries("Sleeping before try refreshing sources again", maxRetriesMultiplier);
         }
       }
@@ -630,7 +634,7 @@ public ReplicationSourceInterface startup() {
         // keep looping in this thread until initialize eventually succeeds,
         // while the server main startup one can go on with its work.
         sourceRunning = false;
-        uncaughtException(t, e, null, null);
+        retryRefreshing(t, e, null, null);
         retryStartup.set(!this.abortOnError);
         do {
           if (retryStartup.get()) {
@@ -641,7 +645,7 @@ public ReplicationSourceInterface startup() {
               initialize();
             } catch (Throwable error) {
               setSourceStartupStatus(false);
-              uncaughtException(t, error, null, null);
+              retryRefreshing(t, error, null, null);
               retryStartup.set(!this.abortOnError);
             }
           }