HBASE-25590 Bulkload replication HFileRefs cannot be cleared in some …

…cases where set exclude-namespace/exclude-table-cfs

hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfig.java

@@ -29,6 +29,8 @@
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.yetus.audience.InterfaceAudience;
 
+import org.apache.hbase.thirdparty.org.apache.commons.collections4.CollectionUtils;
+
 /**
  * A configuration for the replication peer cluster.
  */
@@ -301,6 +303,10 @@ public String toString() {
    * @return true if the table need replicate to the peer cluster
    */
   public boolean needToReplicate(TableName table) {
+    return needToReplicate(table, null);
+  }
+
+  public boolean needToReplicate(TableName table, byte[] family) {
     String namespace = table.getNamespaceAsString();
     if (replicateAllUserTables) {
       // replicate all user tables, but filter by exclude namespaces and table-cfs config
@@ -314,7 +320,8 @@ public boolean needToReplicate(TableName table) {
       Collection<String> cfs = excludeTableCFsMap.get(table);
       // if cfs is null or empty then we can make sure that we do not need to replicate this table,
       // otherwise, we may still need to replicate the table but filter out some families.
-      return cfs != null && !cfs.isEmpty();
+      return cfs != null && !cfs.isEmpty()
+        && (family == null || !cfs.contains(Bytes.toString(family)));
     } else {
       // Not replicate all user tables, so filter by namespaces and table-cfs config
       if (namespaces == null && tableCFsMap == null) {
@@ -325,7 +332,9 @@ public boolean needToReplicate(TableName table) {
       if (namespaces != null && namespaces.contains(namespace)) {
         return true;
       }
-      return tableCFsMap != null && tableCFsMap.containsKey(table);
+      return tableCFsMap != null && tableCFsMap.containsKey(table)
+        && (family == null || CollectionUtils.isEmpty(tableCFsMap.get(table))
+        || tableCFsMap.get(table).contains(Bytes.toString(family)));
     }
   }
 }