HBASE-27405 Fix the replication hfile/log cleaner report that the replication table does not exist (#4811)

Signed-off-by: Duo Zhang <zhangduo@apache.org>

hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationQueueStorage.java

@@ -178,4 +178,10 @@ public interface ReplicationQueueStorage {
    * created hfile references during the call may not be included.
    */
   Set<String> getAllHFileRefs() throws ReplicationException;
+
+  /**
+   * Whether the replication queue table exists.
+   * @return Whether the replication queue table exists
+   */
+  boolean hasData() throws ReplicationException;
 }

hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/TableReplicationQueueStorage.java

@@ -532,4 +532,13 @@ public class TableReplicationQueueStorage implements ReplicationQueueStorage {
       throw new ReplicationException("failed to getAllHFileRefs", e);
     }
   }
+
+  @Override
+  public boolean hasData() throws ReplicationException {
+    try {
+      return conn.getAdmin().getDescriptor(tableName) != null;
+    } catch (IOException e) {
+      throw new ReplicationException("failed to get replication queue table", e);
+    }
+  }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/master/ReplicationLogCleaner.java

@@ -76,6 +76,14 @@ public class ReplicationLogCleaner extends BaseLogCleanerDelegate {
     if (this.getConf() == null) {
       return;
     }
+    try {
+      if (!rpm.getQueueStorage().hasData()) {
+        return;
+      }
+    } catch (ReplicationException e) {
+      LOG.error("Error occurred while executing queueStorage.hasData()", e);
+      return;
+    }
     canFilter = rpm.getReplicationLogCleanerBarrier().start();
     if (canFilter) {
       notFullyDeadServers = getNotFullyDeadServers.get();

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/master/TestReplicationLogCleaner.java

@@ -86,6 +86,7 @@ public class TestReplicationLogCleaner {
     when(rpm.listPeers(null)).thenReturn(new ArrayList<>());
     ReplicationQueueStorage rqs = mock(ReplicationQueueStorage.class);
     when(rpm.getQueueStorage()).thenReturn(rqs);
+    when(rpm.getQueueStorage().hasData()).thenReturn(true);
     when(rqs.listAllQueues()).thenReturn(new ArrayList<>());
     ServerManager sm = mock(ServerManager.class);
     when(services.getServerManager()).thenReturn(sm);